Skip to content

feat: RFC 3507 ICAP client with async, SSL/TLS, and pytest plugin - #20

Merged
CaptainDriftwood merged 99 commits into
masterfrom
feat/rfc3507-icap-client
Jan 11, 2026
Merged

CaptainDriftwood merged 99 commits into
masterfrom
feat/rfc3507-icap-client

Conversation

@CaptainDriftwood

@CaptainDriftwood CaptainDriftwood commented Jan 5, 2026 •

Copy link
Copy Markdown
Owner

Summary

Complete rewrite of the ICAP client with full RFC 3507 compliance, async support, SSL/TLS encryption, preview mode, and comprehensive testing infrastructure including a full-featured pytest plugin with mocking support.

Core Implementation

Sync Client (IcapClient)

  • Full RFC 3507 protocol implementation (OPTIONS, REQMOD, RESPMOD)
  • SSL/TLS support via ssl_context parameter for secure connections
  • Preview mode for efficient scanning of large files
  • Response parsing with Content-Length and chunked transfer encoding
  • Context manager for automatic resource cleanup
  • Custom exception hierarchy (IcapConnectionError, IcapProtocolError, IcapTimeoutError, IcapServerError)
  • Public is_connected property for connection status

Async Client (AsyncIcapClient)

  • Full async/await API with feature parity to sync client
  • SSL/TLS support via ssl_context parameter
  • Preview mode support
  • Async context manager (async with)
  • All ICAP methods: options(), respmod(), reqmod()
  • Convenience methods: scan_bytes(), scan_file(), scan_stream()
  • Proper timeout handling with asyncio.wait_for()
  • Public is_connected property

Shared Protocol Module (_protocol.py)

  • Base class with shared constants and utilities
  • Reduces code duplication between sync and async clients

SSL/TLS Support

import ssl
from pycap import IcapClient

ssl_context = ssl.create_default_context()
with IcapClient('icap.example.com', 11344, ssl_context=ssl_context) as client:
    response = client.scan_file('/path/to/file.pdf')

Preview Mode

Efficient scanning by sending only the first N bytes initially:

with IcapClient('localhost', 1344) as client:
    # Server will request full content only if needed
    response = client.respmod(large_content, service='avscan', preview=1024)

Convenience Methods

  • scan_file(filepath) - Scan files by path (accepts Path or str)
  • scan_stream(stream, chunk_size) - Scan file-like objects with optional chunked streaming
  • scan_bytes(data) - Scan raw bytes directly

Pytest Plugin (pytest_pycap)

Live Client Fixtures

  • icap_client - Sync client with automatic cleanup
  • async_icap_client - Async client with automatic cleanup
  • icap_service_config - Default configuration
  • sample_clean_content - Test content
  • sample_file - Temporary test file
  • @pytest.mark.icap(host, port, timeout, ssl_context) marker for configuration

Mock Client Fixtures (no server required)

  • mock_icap_client - Mock client with default clean responses
  • mock_async_icap_client - Async mock client
  • mock_icap_client_virus - Mock configured to detect viruses
  • mock_icap_client_timeout - Mock that raises IcapTimeoutError
  • mock_icap_client_connection_error - Mock that raises IcapConnectionError

Response Fixtures

  • icap_response_builder - Factory for building custom responses
  • icap_response_clean - Pre-built 204 No Modification
  • icap_response_virus - Pre-built virus detection response
  • icap_response_options - Pre-built OPTIONS response
  • icap_response_error - Pre-built 500 error response

IcapResponseBuilder

Fluent builder for creating test responses:

from pytest_pycap import IcapResponseBuilder

response = IcapResponseBuilder().virus("Trojan.Generic").build()
response = IcapResponseBuilder().clean().build()
response = IcapResponseBuilder().error(503, "Service Unavailable").build()

MockIcapClient

Full mock implementation with advanced features for testing without network I/O:

Response Configuration:

from pytest_pycap import MockIcapClient, IcapResponseBuilder

client = MockIcapClient()

# Single response (reusable)
client.on_respmod(IcapResponseBuilder().virus("Trojan.Gen").build())

# Response sequence (consumed in order)
client.on_respmod(
    IcapResponseBuilder().clean().build(),
    IcapResponseBuilder().virus("Trojan").build(),
    IcapResponseBuilder().clean().build(),
)

Dynamic Callbacks:

def eicar_detector(data: bytes, **kwargs):
    if b"EICAR" in data:
        return IcapResponseBuilder().virus("EICAR-Test").build()
    return IcapResponseBuilder().clean().build()

client.on_respmod(callback=eicar_detector)

Content Matchers:

client.when(filename_matches=r".*\.exe$").respond(
    IcapResponseBuilder().virus("Blocked.Exe").build()
)
client.when(data_contains=b"EICAR").respond(
    IcapResponseBuilder().virus("EICAR-Test").build()
)

Rich Call Inspection:

client.scan_bytes(b"content", filename="test.txt")
call = client.last_call
call.data        # b"content"
call.filename    # "test.txt"
call.was_clean   # True
call.matched_by  # "default"

Strict Mode:

client = MockIcapClient(strict=True)
client.on_respmod(r1, r2)
client.scan_bytes(b"file1")
client.scan_bytes(b"file2")
client.assert_all_responses_used()  # Passes - all consumed

Assertions:

client.assert_called("scan_bytes", times=1)
client.assert_scanned(b"content")
client.assert_called_with("scan_bytes", filename="test.txt")
client.assert_any_call("scan_bytes", data=b"specific")
client.assert_called_in_order(["options", "scan_bytes"])

icap_mock Marker

Declarative mock configuration with stacked response support:

# Simple configuration
@pytest.mark.icap_mock(response="virus", virus_name="Trojan.Custom")
def test_virus_detection(icap_mock):
    response = icap_mock.scan_bytes(b"malware")
    assert response.headers["X-Virus-ID"] == "Trojan.Custom"

# Stacked responses (consumed top-to-bottom)
@pytest.mark.icap_response("clean")
@pytest.mark.icap_response("virus")
@pytest.mark.icap_response("clean")
def test_sequence(icap_mock):
    r1 = icap_mock.scan_bytes(b"file1")  # clean
    r2 = icap_mock.scan_bytes(b"file2")  # virus
    r3 = icap_mock.scan_bytes(b"file3")  # clean

# Strict mode (fails if configured responses not consumed)
@pytest.mark.icap_mock(strict=True)
@pytest.mark.icap_response("clean")
@pytest.mark.icap_response("virus")
def test_strict(icap_mock):
    icap_mock.scan_bytes(b"file1")
    icap_mock.scan_bytes(b"file2")
    # Teardown validates all responses were used

Testing Features

  • Docker availability check - gracefully skips tests if Docker isn't running
  • Automatic plugin registration via entry points
  • Comprehensive pytester tests for plugin validation

Docker Testing Infrastructure

Complete c-icap + ClamAV + squidclamav stack:

  • Multi-stage Dockerfile (65% size reduction: 602MB → 212MB)
  • TLS-enabled ICAP server on port 11344
  • docker-compose.yml for orchestration
  • EICAR virus detection tests
  • SSL/TLS integration tests
  • Integration tests with testcontainers-python

Examples

  • examples/basic_example.py - Basic usage patterns
  • examples/integration_test.py - Docker integration testing
  • examples/lambda_handler.py - AWS Lambda handler for S3 virus scanning

Modern Python Packaging

  • pyproject.toml with Hatchling build backend
  • Python 3.8 - 3.14 support
  • [dependency-groups] for dev dependencies (PEP 735)
  • uv.lock for reproducible builds
  • Ruff for linting/formatting
  • Type hints throughout
  • Keywords for PyPI discoverability
  • MIT LICENSE

CI/CD Workflows

  • test.yml - Matrix testing across Python 3.8-3.14
  • lint.yml - Ruff linting and formatting checks
  • typecheck.yml - Type checking with ty
  • codeql.yml - CodeQL security scanning

Multi-Version Testing

  • noxfile.py with uv backend for fast environment creation
  • Tests across Python 3.8, 3.9, 3.10, 3.11, 3.12, 3.13, 3.14
  • just nox recipes for easy invocation

Usage

# Sync client
from pycap import IcapClient

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

# Async client
from pycap import AsyncIcapClient

async with AsyncIcapClient('localhost', 1344) as client:
    response = await client.scan_bytes(content)
    print(f"Status: {response.status_code}")

# With SSL/TLS
import ssl
ssl_ctx = ssl.create_default_context()
with IcapClient('secure.example.com', 11344, ssl_context=ssl_ctx) as client:
    response = client.scan_bytes(content)

Testing

# Unit tests
uv run pytest -m "not integration"

# Integration tests (requires Docker)
uv run pytest -m integration

# Or use just
just test
just test-integration

# Multi-version testing
just nox

Issues Closed

CaptainDriftwood and others added 30 commits September 20, 2020 15:28
…, refactor scan methods

Co-authored-by: CaptainDriftwood <[email protected]>
…pport

- Wire up custom exceptions (IcapConnectionError, IcapTimeoutError,
  IcapProtocolError, IcapServerError) throughout the client
- Add null socket check in _send_and_receive to prevent AttributeError
- Wrap socket operations with try/except to raise appropriate exceptions
- Add chunked streaming support via scan_stream(chunk_size=N) parameter
- Add _scan_stream_chunked() for memory-efficient large file scanning
- Extract _receive_response() helper for response parsing
ICAP Protocol Improvements:
- Implement chunked transfer encoding for RESPMOD body (RFC 3507 requirement)
- Add chunked response parsing via _read_chunked_body()
- Fix encapsulated header offset calculation for RESPMOD

Testing Improvements:
- Replace hardcoded 30s sleep with smart polling (wait_for_icap_service)
- Add pytest-timeout for graceful test failure on hangs
- Add separate integration test workflow with manual trigger
- Migrate CI to use uv for dependency management

Documentation:
- Add comprehensive ICAP protocol explanation with architecture diagram
- Document c-icap, SquidClamav, and ClamAV with official links
- Update project structure (setup.py → pyproject.toml, add uv.lock)
- Add Python 3.13 to supported versions
The c-icap-mod-virus-scan and squidclamav packages don't exist in
Ubuntu 22.04 repositories. This commit fixes the Docker build by:

- Building squidclamav v7.2 from source instead of using non-existent packages
- Creating architecture-independent symlink for squidclamav.so module
- Changing invalid LogLevel directive to DebugLevel in c-icap.conf
- Adding ServicesDir directive to point to the correct module location
- Making freshclam blocking to ensure virus definitions are loaded before clamd starts
- Increasing retry attempts and adding better startup diagnostics
Add justfile with common development tasks:
- install: Install dependencies with uv
- check/ci: Run all checks (lint, typecheck, test)
- test/test-integration/test-all: Run pytest with various filters
- lint/lint-fix: Run ruff linter
- fmt/fmt-check: Run ruff formatter
- typecheck: Run pyright
- docker-build/docker-up/docker-down/docker-logs: Docker operations
- build: Build package
- clean: Remove build artifacts
… docs

- Add note about PyPI name collision in installation section
- Add error handling section with exception examples
- Add EICAR test string example for virus detection testing
- Fix docker compose commands to use correct path
- Document just command runner as alternative to raw uv commands
- Remove contributing section
- Implement multi-stage Dockerfile separating build and runtime stages
- Remove build-essential and other compile-time dependencies from final image
- Add ca-certificates and clamav-freshclam for proper SSL and virus updates
- Use --no-install-recommends to minimize runtime dependencies
- Remove obsolete 'version' key from docker-compose.yml
- Increase healthcheck retries and add start_period for slower container startup
- Add conftest.py to apply 300s timeout to integration tests

Image size reduced from 602MB to 212MB.
- Add isort configuration with known-first-party for pycap
- Update justfile fmt/fmt-check recipes to include import sorting
- Apply ruff formatting fixes to icap.py
Filter the @wait_container_is_ready deprecation warning from testcontainers.
The recommended replacement (HealthcheckWaitStrategy) has a bug with
DockerCompose containers - it accesses wrapped.attrs which doesn't exist
on ComposeContainer objects.

Document the issue in test_integration.py with references to:
- testcontainers/testcontainers-python#241
- testcontainers/testcontainers-python#144

Our ICAP-level polling approach is actually more robust as it verifies
the protocol is responding, not just that the container is healthy.
Add tests for all mock functionality:
- IcapResponseBuilder (14 tests)
- MockCall (1 test)
- MockIcapClient (21 tests)
- MockAsyncIcapClient (4 tests)
- Mock fixtures (5 tests)
- Response fixtures (5 tests)
- icap_mock marker (8 tests)

Also update plugin.py to export all new fixtures.
Add pytest marker definitions to prevent "unknown marker" warnings
when using --strict-markers.
Expand pytester test coverage from 9 to 27 tests:

Mock fixture tests (6):
- mock_icap_client, mock_async_icap_client
- mock_icap_client_virus, mock_icap_client_timeout
- mock_icap_client_connection_error

Response fixture tests (4):
- icap_response_builder, icap_response_clean
- icap_response_virus, icap_response_error

icap_mock marker tests (4):
- Marker registration, virus response
- Exception raising, per-method config

Mock client usage tests (5):
- Call recording, custom response configuration
- Context manager, fluent builder API, exports
Add comprehensive documentation for the pytest plugin:

- Live client fixtures (icap_client, async_icap_client)
- Mock client fixtures (mock_icap_client, mock_async_icap_client, etc.)
- Response fixtures (icap_response_clean, icap_response_virus, etc.)
- IcapResponseBuilder fluent API with all methods
- MockIcapClient configuration and assertion methods
- @pytest.mark.icap_mock marker with all parameters

Update table of contents to reflect new sections.
Pin all GitHub Actions to immutable commit SHAs instead of version tags
to mitigate supply chain attacks like CVE-2025-30066 (tj-actions incident).

Actions pinned:
- actions/checkout@v4
- astral-sh/setup-uv@v5
- github/codeql-action/init@v4
- github/codeql-action/analyze@v4
- Fix README.md pytest command and project structure
- Add comprehensive docstrings to IcapResponse with attributes and examples
- Document exception hierarchy with usage scenarios
- Enhance IcapClient/AsyncIcapClient with API overview and thread safety
- Improve pytest_pycap fixtures documentation with categories and guidance
- Add detailed IcapResponseBuilder method documentation
- Enhance MockIcapClient/MockAsyncIcapClient with configuration and assertion API
Replace the redundant protocol method listing with highlights of
the library's key features: sync/async clients, high-level scanning
API, TLS support, pytest plugin, and zero dependencies.
Add ability to queue multiple responses that are consumed in FIFO order.
This enables testing scenarios where successive calls need different
responses (e.g., first scan clean, second scan detects virus).

Changes:
- on_respmod(), on_options(), on_reqmod() now accept variadic *responses
- Add _response_queues and _queue_active tracking to MockIcapClient
- Add _get_method_response() for queue-aware response resolution
- Add MockResponseExhaustedError when queue is depleted
- Add reset_responses() to clear queues and restore defaults
- Mirror all changes in MockAsyncIcapClient
- Add 11 comprehensive tests for sequence behavior
…port

Add declarative response sequences via stacked markers on test functions.
Markers are processed in visual top-to-bottom order and support preset
strings ("clean", "virus", "error") or IcapResponse objects.

Changes:
- Register icap_response marker in pytest_configure()
- Add _resolve_marker_response() helper to convert markers to responses
- Update icap_mock fixture to collect and apply stacked markers
- Add 9 pytester-based integration tests for marker functionality

Supports:
- @pytest.mark.icap_response("clean")
- @pytest.mark.icap_response("virus", virus_name="Trojan.Gen")
- @pytest.mark.icap_response("error", code=503, message="Unavailable")
- @pytest.mark.icap_response(response_object)
Add callback= parameter to on_respmod() for dynamic response logic based
on request content, filename, or service. Callbacks have highest priority
in response resolution (before queues and defaults).

Changes:
- Add ResponseCallback and AsyncResponseCallback Protocol classes
- Add _callbacks dict to MockIcapClient for per-method callbacks
- Add callback= parameter to on_respmod() method
- Update _get_method_response() to invoke callbacks first
- Add _get_method_response_async() for async callback support
- Support both sync and async callbacks in MockAsyncIcapClient
- Export callback protocols from pytest_pycap package
- Add 13 unit tests for callback behavior

Example usage:
    def eicar_detector(data: bytes, **kwargs) -> IcapResponse:
        if b"EICAR" in data:
            return IcapResponseBuilder().virus("EICAR-Test").build()
        return IcapResponseBuilder().clean().build()

    client.on_respmod(callback=eicar_detector)
…responses

Add when().respond() API for configuring conditional mock responses
based on service name, filename, or content patterns. Matchers are
checked first in response resolution, taking priority over callbacks,
queues, and defaults.

Features:
- ResponseMatcher dataclass for storing match criteria
- MatcherBuilder class for fluent API via when()
- Support for exact filename, regex patterns, and data_contains
- Combined criteria with AND logic
- times= parameter for limited-use matchers
- Method chaining for registering multiple matchers
- Async client support via _get_method_response_async()

Example:
  client.when(filename="malware.exe").respond(virus_response)
  client.when(data_contains=b"EICAR").respond(virus_response, times=2)
… assertions

- Enhance MockCall dataclass with response, exception, matched_by, call_index fields
- Add convenience properties: data, filename, service, succeeded, was_clean, was_virus
- Improve MockCall.__repr__ for rich debugging output
- Add call query methods: first_call, last_call, last_scan_call, get_calls, get_scan_calls
- Add call statistics: call_count, call_counts_by_method
- Add enhanced assertions: assert_called_with, assert_any_call, assert_called_in_order,
  assert_scanned_file, assert_scanned_with_filename
- Mirror all changes in MockAsyncIcapClient
- Add 56 unit tests for Phase 4 functionality
Add strict mode to MockIcapClient that validates all configured
responses were consumed during tests. This helps catch test setup
that doesn't match actual test behavior.

- Add strict parameter to MockIcapClient.__init__
- Track queue sizes, callback usage, and matcher triggers
- Add assert_all_responses_used() method for explicit validation
- Add @pytest.mark.icap_mock(strict=True) marker support
- Automatic teardown validation when strict mode enabled via marker
- Add 20 new tests for strict mode functionality
- Update MockIcapClient class docstring with comprehensive overview
  of all 6 capabilities (response config, sequences, callbacks,
  matchers, call recording, strict mode)
- Update mock.py module docstring with key features and examples
- Expand README MockIcapClient section with examples for all features
- Add stacked response markers documentation to README
- Add strict mode parameter to icap_mock marker docs
- Fix socket resource leak on connection error (BUG-001)
  - Close socket in exception handlers before setting to None
- Fix async writer not properly closed on error (BUG-002)
  - Call writer.close() before setting to None in error handlers
- Fix silent data truncation on incomplete response (BUG-003)
  - Validate bytes_read == content_length, raise IcapProtocolError if mismatch
- Fix silent truncation on malformed chunk size (BUG-004)
  - Raise IcapProtocolError instead of returning partial data
- Fix connection state not updated on socket errors (BUG-005)
  - Set _connected = False when socket errors occur in _receive_response()
- Fix unhandled ValueError in Content-Length parsing (BUG-006)
  - Wrap int() in try/except, raise IcapProtocolError with context
- Add preview size validation (BUG-007)
  - Validate preview parameter is positive integer
Add 29 unit tests covering error handling and edge cases:
- IcapProtocolError for malformed responses, invalid Content-Length, invalid chunk sizes
- IcapServerError for 5xx responses (500, 502, 503, 505)
- _read_chunked_body() with various inputs and edge cases
- Connection lifecycle (reconnect, idempotent connect, safe disconnect)
- scan_file() with non-existent file (FileNotFoundError)
- Preview size validation (ValueError for zero/negative values)
- Response parsing edge cases (empty body, Content-Length: 0, multi-word status)
Add `from None` to re-raised exceptions in except clauses to satisfy
ruff B904 rule and properly distinguish exceptions from error handling.
- Fix silent chunked body truncation: raise IcapProtocolError when
  connection closes before chunked body complete (HIGH severity)
- Wrap stream.read() errors: catch IOError/OSError and re-raise as
  IcapProtocolError with context (MEDIUM severity)
- Add chunk_size parameter to AsyncIcapClient.scan_stream() for API
  parity with sync client (LOW severity)
- Add 5 new tests covering these fixes
Add 36 new tests covering:
- _send_with_preview() tests (timeout, connection error, 100 Continue flow)
- _receive_response() tests (sync and async)
- _read_chunked_body() tests (sync and async)
- _scan_stream_chunked() tests
- _iter_chunks() tests (sync and async)
- reqmod() and options() public API tests
- Real socket timeout and connection error scenarios
- Sync/async API parity verification tests

All tests pass (287 total, 26 deselected for integration/SSL).
- Add pytest-mock dependency for cleaner test mocking
- Convert 6 tests from patch/patch.object to mocker.patch:
  - test_error_handling.py: 3 socket patching tests
  - test_client_methods.py: 3 auto-connect tests
- Remove unused patch imports

pytest-mock provides:
- Automatic cleanup (no context managers needed)
- Single consistent API via mocker fixture
- Better integration with pytest ecosystem
- Add 'slow' marker definition to pyproject.toml for future use
- Mock socket connections in test_client_methods.py to avoid real
  network calls that cause slow/flaky tests
- Mock asyncio.open_connection in test_async_integration.py error tests
- Remove section comment headers (TEST-XXX) from test_client_methods.py
- Add comment explaining setuptools dependency for PyCharm
- Add 'docker' marker to pyproject.toml for Docker-dependent tests
- Add reqmod() integration tests for sync and async clients
- Add async scan_stream() integration tests
- Add async respmod() with preview tests (small/large/eicar)
- Add connection reuse tests for sync and async clients
- Add SSL tests for scan_file, scan_stream, and async respmod
- Mark all Docker-dependent tests with @pytest.mark.docker
- Remove redundant comments from pycap/ library code
@CaptainDriftwood CaptainDriftwood changed the title feat: complete ICAP client with sync/async, SSL/TLS, preview mode, pytest plugin, and Docker testing feat: RFC 3507 ICAP client with async, SSL/TLS, and pytest plugin Jan 11, 2026
@CaptainDriftwood CaptainDriftwood added the documentation Improvements or additions to documentation label Jan 11, 2026
@CaptainDriftwood
CaptainDriftwood merged commit 501b13e into master Jan 11, 2026
9 of 10 checks passed
@CaptainDriftwood
CaptainDriftwood deleted the feat/rfc3507-icap-client branch January 11, 2026 23:29
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

documentation Improvements or additions to documentation enhancement New feature or request

Projects

None yet

2 participants