[client-python] feat(signature): first US for ExpectationSignature (#206)

[Kakudou]

That's the first US to introduce the SignatureExpectation for the injectors.

Proposed changes

SignatureManager

Unified signature lifecycle for OpenAEV injectors: compile pre-execution signatures, merge post-execution results, and ship structured output to the backend.

Architecture

flowchart LR
    subgraph pyoaev/signatures
        SM[SignatureManager]
        M[models.py]
        CFG["InjectorConfig\n(Network / Cloud / External)"]
    end
    subgraph pyoaev/apis
        API[SignatureApiManager]
    end
    subgraph Backend
        CB["/api/injects/{id}/callback"]
    end

    CFG -->|typed input| SM
    SM -->|compile_pre/post| M
    SM -->|send_signatures| API
    API -->|callback\nretry + chunk| CB

Loading

Injector configs (models.py) are the typed contract: one config = one signature row.
SignatureManager owns the domain logic (compile, merge, resolve IP).
SignatureApiManager owns the transport (validation, chunking, retry).

Quick Start

from pyoaev import OpenAEV
from pyoaev.signatures import (
    SignatureManager,
    NetworkInjectorConfig,
    build_network_configs,
)

client = OpenAEV(url="https://openaev.example.com", token="my-token")
sm = SignatureManager(client)

# 1. Build typed injector configs (one per distinct target asset)
configs = build_network_configs(["10.0.0.1", "2001:db8::1", "target.example.com"])
# or hand-build them: NetworkInjectorConfig(target_ipv4="10.0.0.1")
# or build a list: [NetworkInjectorConfig(target_ipv4="10.0.0.1"), NetworkInjectorConfig(target_ipv6="2001:db8::1"), NetworkInjectorConfig(target_domain="target.example.com")]
  
# 2. Compile pre-execution signatures (category is carried by the config type)
pre = sm.compile_pre_execution_signatures(config=configs)

# 3. Run your tool...

# 4. Compile post-execution signatures
post = sm.compile_post_execution_signatures(pre, tool_output)

# 5. Build the wire payload
target_meta = {"agent": agent_id, "asset": asset_id, "asset_group": group_id}
payload = sm.build_payload(post, target_meta, expectation_type="DETECTION")

# 6. Send to backend
sm.send_signatures(inject_id="abc-123", phase="execution_complete", signatures=payload)

Injector configs

The category is encoded in the config type. Pass a single config for a single-target inject,
or a homogeneous list for a multi-target inject. Mixing config types in a single call is rejected.

Config	Required fields	Optional fields	Use case
NetworkInjectorConfig	target_ipv4 / target_ipv6 / target_hostname	—	Nuclei, Nmap, NetExec
CloudInjectorConfig	cloud_provider, cloud_account_id, cloud_region	target_service	Prowler, Stratus
ExternalInjectorConfig	query	target_ipv4, target_hostname	Shodan

InjectorConfig is the union type: NetworkInjectorConfig | CloudInjectorConfig | ExternalInjectorConfig.
SignatureManager adds start_time automatically (plus source_ipv4 / source_ipv6 for network).

Network

from pyoaev.signatures import NetworkInjectorConfig

# One distinct asset per config, never mix identities on the same target
cfg = NetworkInjectorConfig(target_ipv4="10.0.0.1")
cfg = NetworkInjectorConfig(target_ipv6="2001:db8::1")
cfg = NetworkInjectorConfig(target_hostname="api.example.com")

# Multi-target inject
configs = [
    NetworkInjectorConfig(target_ipv4="10.0.0.1"),
    NetworkInjectorConfig(target_hostname="api.example.com"),
]
pre = sm.compile_pre_execution_signatures(config=configs)
# -> list of dicts, one per target, all sharing the same source_ipv4 / start_time

####### Network builder

build_network_configs(targets) turns a heterogeneous list of strings, dicts, or already-typed
NetworkInjectorConfig into a clean list of typed configs. Strings are auto-classified into
IPv4 / IPv6 / hostname via the stdlib ipaddress module. Each input is treated as one distinct
asset — a target never mixes identities.

from pyoaev.signatures import build_network_configs

build_network_configs(["10.0.0.1", "2001:db8::1", "web.example.com"])
# -> [NetworkInjectorConfig(target_ipv4="10.0.0.1"),
#     NetworkInjectorConfig(target_ipv6="2001:db8::1"),
#     NetworkInjectorConfig(target_hostname="web.example.com")]

# dicts also work and are validated
build_network_configs([{"target_ipv4": "10.0.0.1"}])

Cloud

from pyoaev.signatures import CloudInjectorConfig

cfg = CloudInjectorConfig(
    cloud_provider="aws",
    cloud_account_id="123456789012",
    cloud_region="eu-west-1",
    target_service="ec2",  # optional
)

# Multi-region: one config per region
configs = [
    CloudInjectorConfig(cloud_provider="aws", cloud_account_id="123456789012", cloud_region=r)
    for r in ("us-east-1", "eu-west-1", "ap-southeast-1")
]
pre = sm.compile_pre_execution_signatures(config=configs)

External

from pyoaev.signatures import ExternalInjectorConfig

cfg = ExternalInjectorConfig(
    query="port:22 os:linux",
    target_ipv4="203.0.113.5",      # optional
    target_hostname="ssh.example.com",  # optional
)
pre = sm.compile_pre_execution_signatures(config=cfg)

Compiled output shapes

compile_pre_execution_signatures returns a single flat dict for one config, or a list of dicts
for a list of configs. None fields are stripped.

# Network single target
{
    "start_time": "2024-06-26T06:06:06Z",
    "source_ipv4": "172.17.0.2",
    "target_ipv4": "10.0.0.1",
}

# Cloud single region
{
    "start_time": "2024-06-26T06:06:06Z",
    "cloud_provider": "aws",
    "cloud_account_id": "123456789012",
    "cloud_region": "eu-west-1",
    "target_service": "ec2",
}

# External
{
    "start_time": "2024-06-26T06:06:06Z",
    "target_ipv4": "203.0.113.5",
    "query": "port:22 os:linux",
}

compile_post_execution_signatures(pre, tool_output) preserves the input shape (dict in, dict out;
list in, list out) and adds end_time, execution_status, and optional partial_results.

# tool_output examples → execution_status
{}                                                  # -> "success"
{"status": "partial"}                               # -> "partial"
{"error_info": {"exit_code": 1}}                    # -> "failed"
{"timeout_info": {"partial_results": ["host-a"]}}   # -> "timeout"

Anything in tool_output["extra_signatures"] is merged into the final dict verbatim, useful for
injector-specific fields like parent_process_name or custom signal types.

Failure modes

Trigger	Result
Empty list passed to compile_pre_execution_signatures	ValueError
List mixing config types (e.g. Network + Cloud)	ValueError
NetworkInjectorConfig with zero or more than one identity field	ValidationError
build_network_configs item that's neither str, dict, nor a NetworkInjectorConfig	TypeError
Malformed tool_output in post-execution	OpenAEVError

Lifecycle Flow

sequenceDiagram
    participant Injector
    participant SM as SignatureManager
    participant API as SignatureApiManager
    participant Backend

    Injector->>SM: compile_pre_execution_signatures(config)
    SM-->>Injector: pre_signatures dict/list

    Note over Injector: Tool executes...

    Injector->>SM: compile_post_execution_signatures(pre, tool_output)
    SM-->>Injector: merged signatures

    Injector->>SM: build_payload(post, target_meta, expectation_type)
    SM-->>Injector: nested wire payload

    Injector->>SM: send_signatures(inject_id, phase, signatures)
    SM->>API: send_signatures(inject_id, phase, signatures)
    API->>API: validate + normalize + chunk if needed
    API->>Backend: POST /api/injects/{id}/callback
    Backend-->>API: 200/202

Loading

Transport Behaviour

• Auto-chunking: payloads exceeding max_payload_size (default 1 MiB) are split by target and sent sequentially with chunk_index / total_chunks metadata.
• Retry: 5xx errors trigger up to 3 retries with exponential backoff (1s, 2s, 4s).
• No retry on 4xx: client errors raise SignatureTransmissionError immediately.

Wire Format

Payloads follow the nested schema expected by the callback endpoint:

{
  "phase": "execution_complete",
  "expectation_signature": {
    "targets": [
      {
        "signature_target": { "agent": "...", "asset": "...", "asset_group": "..." },
        "signature_values": [
          {
            "expectation_type": "DETECTION",
            "values": [
              { "signature_type": "source_ipv4", "signature_value": "172.17.0.2" },
              { "signature_type": "target_ipv4", "signature_value": "10.0.0.1" },
              { "signature_type": "start_time", "signature_value": "2024-06-26T06:06:06Z" },
              { "signature_type": "end_time", "signature_value": "2024-06-26T06:06:09Z" },
              { "signature_type": "execution_status", "signature_value": "success" }
            ]
          }
        ]
      }
    ]
  }
}

Known signature_type labels live in pyoaev.signatures.SignatureTypes
(source_ipv4_address, target_hostname_address, cloud_region, query, ...). The wire format
itself accepts any string, so injectors are free to add custom types via tool_output.extra_signatures.

Utility

ip = sm.resolve_container_ip()  # "172.17.0.2" or "unknown" with a warning

Resolution strategy: CONTAINER_IP env var > socket.gethostbyname > hostname -i > "unknown".
The result is cached for the lifetime of the manager and IPv6 is sniffed best-effort alongside.

Related issues

• Add new ContractOutputType: ExpectationSignature #206

Checklist

• I consider the submitted work as finished
• I tested the code for its functionality
• I wrote test cases for the relevant uses case
• I added/update the relevant documentation (either on github or on notion)
• Where necessary I refactored code to improve the overall quality
• For bug fix -> I implemented a test that covers the bug

Further comments

[Kakudou]

Initial message edited to reflect the new behavior based on review/usages.
The way we defined the inject_config and the category was tedious to use, so instead i've created 3new models:
NetworkInjectorConfig, CloudInjectorConfig and ExternalInjectorConfig the usage of one of them (can't be mixin) define the type of injector.

Also as for the NetworkInjectorConfig, we can use a builder to quickly create them from a list of targets (from Targets.extract_targets()>.targets per example) or from a dict:

build_network_configs(["10.0.0.1", "2001:db8::1", "web.example.com"])
build_network_configs([{"target_ipv4": "10.0.0.1"}])

[guzmud]

Quick question @Kakudou about adc201b : wouldn't it make sense to update main with a different PR (CI-oriented to use the dev requirements as you suggested) and rebase this one? (now that we are in rolling release, we can use main like that)

[guzmud]

Should we use a pydantic model_validator (eventually a mode=before) to ensure there is at least one but only one of those three options?

Something like

@model_validator(mode='before')
def check_one(cls, data):
    assert sum(value != None for key, value in data.items() if key in ['target_ipv4', 'target_ipv6', 'target_hostname']) == 1
    return data