Metadata-Version: 2.5
Name: faulty
Version: 0.1.3
Summary: Lightweight cross-platform Asyncio TCP Chaos Proxy for network failure simulation.
Project-URL: Homepage, https://github.com/ken1zy/faulty
Project-URL: Repository, https://github.com/ken1zy/faulty
Author: Timofey Botalov
License: MIT
License-File: LICENSE
Keywords: asyncio,chaos-engineering,network,proxy,tcp-proxy,testing
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Testing
Requires-Python: >=3.10
Requires-Dist: rich>=13.0.0
Requires-Dist: typer>=0.9.0
Description-Content-Type: text/markdown

# Faulty

Asynchronous TCP chaos proxy for emulating network problems during local development and in automated tests.

[![Tests](https://github.com/ken1zy/faulty/actions/workflows/tests.yml/badge.svg)](https://github.com/USER/REPO/actions/workflows/tests.yml)
[![Python 3.8+](https://img.shields.io/badge/python-3.8+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

## Description

Faulty is a lightweight TCP proxy based on `asyncio` that injects network failures between client and server. Runs in user space without root privileges, `tc`, `iptables`, or kernel modifications. Supports Linux, macOS, and Windows.

**Key features:**

- Pure Python implementation using asynchronous generators
- Separate pipelines for ingress (download) and egress (upload)
- Middleware architecture for composing failure injection rules
- No system privileges required

## Installation

```bash
pip install faulty
```

## Basic Usage

```python
import asyncio
from faulty.core import FaultyProxy
from faulty.middlewares import LatencyMiddleware, DropMiddleware

async def main():
    proxy = FaultyProxy(
        listen_host="127.0.0.1",
        listen_port=8080,
        target_host="api.example.com",
        target_port=443
    )
    
    # Add latency on download (ingress)
    proxy.ingress_pipeline.middlewares.append(
        LatencyMiddleware(min_ms=50, max_ms=150)
    )
    
    # Drop 10% of packets on upload (egress)
    proxy.egress_pipeline.middlewares.append(
        DropMiddleware(drop_rate=0.1)
    )
    
    await proxy.start()

asyncio.run(main())
```

Configure your application to connect to `127.0.0.1:8080` instead of `api.example.com:443`.

## CLI

Faulty can be run from the command line without writing code:

```bash
# Basic startup
faulty start 8080 httpbin.org:80

# With 100-300 ms latency in all directions
faulty start 8080 httpbin.org:80 --delay 100-300

# Limit download speed to 64 KB/s
faulty start 8080 httpbin.org:80 --throttle-down 64kb

# Combination of multiple rules
faulty start 127.0.0.1:8080 api.example.com:443 \
    --delay-down 200 \
    --throttle-up 128kb \
    --drop 0.05 \
    --corrupt-down 0.01
```

**Available flags:**

- `--delay`, `-d` — latency for all directions (ms): `200` or `200-500`
- `--drop`, `-p` — packet loss probability (0.0–1.0)
- `--throttle`, `-t` — speed limit: `64kb`, `1mb`
- `--corrupt`, `-c` — byte corruption probability (0.0–1.0)

Add suffix `-down` (ingress) or `-up` (egress) to apply rules in one direction only:

- `--delay-down`, `--delay-up`
- `--drop-down`, `--drop-up`
- `--throttle-down`, `--throttle-up`
- `--corrupt-down`, `--corrupt-up`

## pytest Integration

```python
import pytest
from faulty.core import FaultyProxy
from faulty.middlewares import ThrottleMiddleware

@pytest.mark.asyncio
async def test_slow_network_behavior():
    async with FaultyProxy("127.0.0.1", 9000, "localhost", 5432) as proxy:
        # Limit download speed to 10 KB/s
        proxy.ingress_pipeline.middlewares.append(
            ThrottleMiddleware(bytes_per_sec=10_000)
        )
        
        # Test code connects to 127.0.0.1:9000
        # instead of localhost:5432
        ...
```

## Built-in Middleware

| Middleware | Parameters | Description |
|----------|-----------|----------|
| `LatencyMiddleware` | `min_ms`, `max_ms` | Adds random latency from min to max milliseconds |
| `DropMiddleware` | `drop_rate` | Drops packets with probability `drop_rate` (0.0–1.0) |
| `ThrottleMiddleware` | `bytes_per_sec`, `chunk_size` | Limits throughput to specified bytes/sec |
| `CorruptBytesMiddleware` | `corruption_rate` | Randomly modifies one byte in a packet with given probability |

## Custom Middleware

Implement `BaseMiddleware` and define an asynchronous generator to process byte chunks:

```python
import random
from typing import AsyncGenerator
from faulty.middlewares import BaseMiddleware

class DuplicatePacketMiddleware(BaseMiddleware):
    def __init__(self, duplicate_rate: float):
        self.duplicate_rate = duplicate_rate
    
    async def process(self, chunk: bytes) -> AsyncGenerator[bytes, None]:
        yield chunk
        if random.random() < self.duplicate_rate:
            yield chunk  # Send duplicate
```

Add to the pipeline:

```python
proxy.egress_pipeline.middlewares.append(
    DuplicatePacketMiddleware(duplicate_rate=0.05)
)
```

## Architecture

- **Pipeline**: Sequential chain of middlewares applied to each data chunk
- **Middleware**: Asynchronous generator that accepts bytes and yields 0 or more byte chunks
- **Ingress/Egress**: Separate pipelines for download and upload directions
- **Buffer size**: Configurable via `FaultyProxy(buffer_size=...)` (default 64 KB)

## License

MIT
