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
payload = sm.build_payload(post, expectation_types=["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_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

• feat: 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]

@Kakudou to keep you updated:

when you speak about send_signatures it seems to be limited to a single expectation type, cf. your example

{
  "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" }
            ]
          }
        ]
      }
    ]
  }
}

yet according to the engineering made beforehand (as in US-5), multiple expectation types could be found there

  "expectation_signatures": {
    "DETECTION": {
      "source_ipv4": "172.18.0.5",
      "target_ipv4": "192.168.1.10",
      "target_hostname": "webserver.corp.local",
      "start_time": "2026-04-20T10:00:00Z",
      "end_time": "2026-04-20T10:05:30Z"
    },
    "PREVENTION": {
      "source_ipv4": "172.18.0.5",
      "target_ipv4": "192.168.1.10",
      "start_time": "2026-04-20T10:00:00Z",
      "end_time": "2026-04-20T10:05:30Z"
    },
    "VULNERABILITY": {
      "cves_tested": ["CVE-2023-1234"],
      "cves_found_vulnerable": ["CVE-2023-1234"],
      "target_ipv4": "192.168.1.10"
    }
  },

you anticipated that properly in your models, SignaturePayload expects a list of TargetSignatures indeed but inside those the signature_values are a list of ExpectationSignatureGroup

class TargetSignatures(BaseModel):
  """A target plus everything observed about it, grouped by expectation."""

  model_config = ConfigDict(extra="allow")

  signature_target: SignatureTarget
  signature_values: list[ExpectationSignatureGroup]


class SignaturePayload(BaseModel):
  """Inner ``signatures`` body: a list of targets, nothing else."""

  model_config = ConfigDict(extra="allow")

  targets: list[TargetSignatures]

but this doesn't translate into your function build_payload where expectation_type is a single type

    def build_payload(
        self,
        post_signatures: dict[str, Any] | list[dict[str, Any]],
        targets_meta: dict[str, str] | list[dict[str, str]],
        expectation_type: str = "DETECTION",
    ) -> dict[str, Any]:

leading to a single ExpectationSignatureGroup in your signature target

            targets.append(
                TargetSignatures(
                    signature_target=SignatureTarget(**meta),
                    signature_values=[
                        ExpectationSignatureGroup(
                            expectation_type=expectation_type, values=values
                        )
                    ],
                )
            )

I've made a small commit 8b7848f in order to allow for multiple expectation types in the build_payload.

cc @Megafredo @mariot

[codecov]

Codecov Report

❌ Patch coverage is 86.19792% with 53 lines in your changes missing coverage. Please review.
✅ Project coverage is 73.19%. Comparing base (181cffd) to head (edcd9ac).
⚠️ Report is 10 commits behind head on main.

Files with missing lines	Patch %	Lines
pyoaev/signatures/signature_manager.py	77.41%	28 Missing ⚠️
pyoaev/apis/signature.py	88.46%	15 Missing ⚠️
pyoaev/signatures/models.py	91.07%	10 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #257      +/-   ##
==========================================
+ Coverage   70.34%   73.19%   +2.84%     
==========================================
  Files          49       53       +4     
  Lines        1966     2350     +384     
==========================================
+ Hits         1383     1720     +337     
- Misses        583      630      +47     

Flag	Coverage Δ
connectors	73.19% <86.19%> (+2.84%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[guzmud]

Nota bene for @mariot and @Kakudou !

In order to have a better control regarding extra_signatures and being able to distribute it according to expectations, @Megafredo and me:

• moved extra_signatures outside of OutputTool (focused now more on serializing metadata regarding a tool run)
• changed the design of extra signatures to be attached to 1+ expectation type (since some extra only makes sense for some specific expectation types)
• created a new ExtraSignatureData model to match this new design(with ExtraSignatureData.detection, ExtraSignatureData.prevention, ExtraSignatureData.vulnerability)
• added this new model as an input for the build_payload function
• and used pydantic.JsonValue rather than str in a couple of places to better match the specs

This gives us the ability to spread the extra signatures into the various expectation types while providing a stronger typing/definition for the dev-experience.