# python-icap

> Pure Python ICAP client library with no external dependencies

python-icap is a Python library for communicating with ICAP (Internet Content Adaptation Protocol) servers. It implements RFC 3507 for integrating with servers like c-icap and SquidClamav for antivirus scanning, content filtering, and data loss prevention.

## Key Features

- Pure Python implementation with zero runtime dependencies
- Sync (`IcapClient`) and async (`AsyncIcapClient`) clients with full API parity
- High-level file scanning: `scan_file()`, `scan_bytes()`, `scan_stream()`
- SSL/TLS support with custom certificates and mutual TLS
- Bundled pytest plugin for testing without a live server
- Python 3.8+ support

## Installation

```bash
pip install python-icap
```

## Quick Start

```python
from icap import IcapClient

with IcapClient('localhost', port=1344) as client:
    response = client.scan_file('/path/to/file.pdf')
    if response.is_no_modification:
        print("File is clean")
    else:
        print("Threat detected")
```

Async usage:

```python
import asyncio
from icap import AsyncIcapClient

async def scan():
    async with AsyncIcapClient('localhost') as client:
        response = await client.scan_bytes(b"content")
        print(f"Clean: {response.is_no_modification}")

asyncio.run(scan())
```

## Public API

### Main Classes

Import from `icap`:

- `IcapClient(address, port=1344, timeout=10, ssl_context=None, max_response_size=104857600)` - Synchronous client
- `AsyncIcapClient(address, port=1344, timeout=10.0, ssl_context=None, max_response_size=104857600)` - Async client (max_response_size limits response size to prevent DoS, default 100MB)
- `IcapResponse` - Response object with `status_code`, `headers`, `body`, `is_success`, `is_no_modification`, `encapsulated`, `istag`
- `EncapsulatedParts` - Dataclass for parsed Encapsulated header offsets (`req_hdr`, `req_body`, `res_hdr`, `res_body`, `null_body`, `opt_body`)
- `CaseInsensitiveDict` - Case-insensitive dictionary for ICAP headers (per RFC 3507)

### High-Level Methods (Recommended)

- `scan_file(filepath, service='avscan', chunk_size=0)` - Scan file by path (chunk_size>0 streams large files)
- `scan_bytes(data, service='avscan', filename=None)` - Scan bytes directly
- `scan_stream(stream, service='avscan', filename=None, chunk_size=0)` - Scan file-like objects

### Low-Level Methods

- `options(service)` - Query server capabilities
- `respmod(service, http_request, http_response, headers=None, preview=None)` - Response modification
- `reqmod(service, http_request, http_body=None, headers=None)` - Request modification

### IcapResponse Properties

- `status_code` - ICAP status code (200, 204, etc.)
- `headers` - Case-insensitive header dictionary
- `body` - Response body bytes
- `is_success` - True for 2xx status codes
- `is_no_modification` - True for 204 (content is clean)
- `encapsulated` - Parsed `EncapsulatedParts` from Encapsulated header (or None)
- `istag` - ISTag header value for cache validation (RFC 3507 Section 4.7)

### Connection Management

- `connect()` / `disconnect()` - Manual connection control
- `is_connected` - Property to check connection status
- Context manager support: `with IcapClient(...) as client:`

### Exceptions

Import from `icap.exception`:

- `IcapException` - Base exception class
- `IcapConnectionError` - Connection failures
- `IcapTimeoutError` - Operation timeouts
- `IcapProtocolError` - Malformed responses
- `IcapServerError` - Server 5xx errors

## Pytest Plugin

The bundled pytest plugin provides fixtures for testing ICAP integrations.

### Mock Fixtures (no server required)

- `mock_icap_client` - Returns clean (204) responses
- `mock_async_icap_client` - Async version
- `mock_icap_client_virus` - Returns virus detection responses
- `mock_icap_client_timeout` - Raises `IcapTimeoutError`
- `mock_icap_client_connection_error` - Raises `IcapConnectionError`

### Response Builder

```python
from icap.pytest_plugin import IcapResponseBuilder

response = IcapResponseBuilder().clean().build()  # 204 No Modification
response = IcapResponseBuilder().virus("Trojan.Gen").build()  # Virus detected
response = IcapResponseBuilder().error(503, "Unavailable").build()  # Error
```

### Pre-built Response Fixtures

- `icap_response_clean` - 204 No Modification response
- `icap_response_virus` - Virus detection response (X-Virus-ID: EICAR-Test)
- `icap_response_options` - OPTIONS response with server capabilities
- `icap_response_error` - 500 Server Error response
- `icap_response_builder` - Factory fixture returning `IcapResponseBuilder`

### MockIcapClient

```python
from icap.pytest_plugin import MockIcapClient, IcapResponseBuilder

client = MockIcapClient()
client.on_respmod(IcapResponseBuilder().virus("Trojan").build())
response = client.scan_bytes(b"content")
client.assert_called("scan_bytes", times=1)
```

### Markers

```python
@pytest.mark.icap_mock(response="clean")
def test_clean(icap_mock):
    assert icap_mock.scan_bytes(b"data").is_no_modification

@pytest.mark.icap_mock(response="virus", virus_name="EICAR")
def test_virus(icap_mock):
    assert not icap_mock.scan_bytes(b"data").is_no_modification
```

### Advanced Mock Features

```python
from icap.pytest_plugin import MockIcapClient, IcapResponseBuilder

# Conditional responses based on scanned content
client = MockIcapClient()
client.when(data_contains=b"EICAR").respond(
    IcapResponseBuilder().virus("EICAR-Test").build()
)
client.on_any(IcapResponseBuilder().clean().build())  # fallback for everything else

# Call inspection
response = client.scan_bytes(b"test")
call = client.last_call
assert call.method == "scan_bytes"
assert call.data == b"test"  # or call.kwargs["data"]

# Strict mode (fails if no matching response)
@pytest.mark.icap_mock(strict=True)
def test_strict(icap_mock):
    icap_mock.on_respmod(IcapResponseBuilder().clean().build())
    # MockResponseExhaustedError if called more times than configured
```

## Project Structure

```
src/icap/
├── __init__.py          # Public API exports
├── icap.py              # Synchronous IcapClient
├── async_icap.py        # AsyncIcapClient
├── response.py          # IcapResponse, EncapsulatedParts, CaseInsensitiveDict
├── exception.py         # Custom exceptions
├── _protocol.py         # Shared protocol utilities
└── pytest_plugin/       # Bundled pytest plugin
    ├── __init__.py      # Plugin entry point: fixtures and markers
    ├── builder.py       # IcapResponseBuilder
    ├── mock_client.py   # MockIcapClient (sync)
    ├── mock_async.py    # MockAsyncIcapClient
    ├── matchers.py      # ResponseMatcher, MatcherBuilder
    ├── call_record.py   # MockCall for call inspection
    ├── protocols.py     # ResponseCallback protocols
    └── mock.py          # Re-exports for backward compatibility
```

## Common Service Names

- `"avscan"` or `"srv_clamav"` - ClamAV virus scanning
- `"squidclamav"` - SquidClamav service
- `"echo"` - Echo service (testing)

## Testing

```bash
# Unit tests
pytest -m "not integration"

# Integration tests (requires Docker)
docker compose -f docker/docker-compose.yml up -d
pytest -m integration
```

## EICAR Test String

Standard test string for triggering antivirus detection:

```python
EICAR = b'X5O!P%@AP[4\\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*'
```

## Protocol Reference

- RFC 3507: Internet Content Adaptation Protocol
- Default port: 1344
- Methods: OPTIONS, REQMOD, RESPMOD

## Links

- Repository: https://github.com/CaptainDriftwood/python-icap
- PyPI: https://pypi.org/project/python-icap/
- c-icap: https://c-icap.sourceforge.net/
- SquidClamav: https://squidclamav.darold.net/
- ClamAV: https://www.clamav.net/