Domain Models¶

class Evidence(BaseModel):
    """Verifiable source reference for a finding.

    Attributes:
        source: Where the evidence came from (e.g. ``"nmap scan"``,
            ``"git log a3f91c"``).
        detail: The raw data that substantiates the finding (packet capture
            excerpt, code snippet, API response body, etc.).
    """

    source: str
    detail: str

class RawFinding(BaseModel):
    """Unprocessed output from a scanner plugin before LLM reasoning.

    Attributes:
        plugin: Name of the plugin that produced this finding.
        asset: The target asset (IP, URL, repo path, etc.).
        raw: Raw output string from the underlying tool.
    """

    plugin: str
    asset: str
    raw: str

class Vulnerability(BaseModel):
    """A single security finding with mandatory evidence.

    Attributes:
        id: Unique finding identifier, e.g. ``"CVE-2024-23897"``.
        title: Short human-readable description.
        severity: ``"critical"``, ``"high"``, ``"medium"``, ``"low"``,
            or ``"info"``.
        asset: The target asset this vulnerability was found on.
        evidence: Verifiable source reference - never empty.
        mitre_ttps: ATT&CK technique IDs applicable to this finding.
    """

    id: str
    title: str
    severity: str
    asset: str = ""
    evidence: Evidence
    mitre_ttps: list[str] = []

class AttackPath(BaseModel):
    """A chained exploit sequence from initial access to impact.

    Attributes:
        narrative: Human-readable kill chain description.
        evidence: Source data that substantiates every step in the chain.
        mitre_ttps: Ordered ATT&CK technique IDs for each hop.
        steps: Ordered list of asset/action pairs describing the path.
    """

    narrative: str
    evidence: Evidence
    mitre_ttps: list[str] = []
    steps: list[str] = []

class Brief:
    """Narrative red team report generated from a :class:`Report`."""

    def __init__(self, text: str = "") -> None:
        self._text = text

    def to_markdown(self) -> str:
        """Return the brief as a Markdown string."""
        raise NotImplementedError

    def to_pdf(self) -> bytes:
        """Return the brief as a PDF byte string."""
        raise NotImplementedError

class Report:
    """Full engagement report returned by :meth:`~matadore.Matadore.engage`.

    Attributes:
        surface: All exposed assets discovered during the engagement.
        vulnerabilities: Findings ranked by exploitability, with evidence.
        _audit: Internal audit log for this engagement.
    """

    def __init__(self, engagement_id: str = "") -> None:
        self.surface: list[str] = []
        self.vulnerabilities: list[Vulnerability] = []
        self._audit = AuditLog(engagement_id=engagement_id)

    def summary(self) -> str:
        """Return a one-paragraph executive summary of the engagement."""
        raise NotImplementedError

    def attack_paths(self) -> list[AttackPath]:
        """Return chained exploit sequences ordered by likelihood."""
        raise NotImplementedError

    def entry_points(self) -> list[str]:
        """Return the most likely initial access vectors."""
        raise NotImplementedError

    def shadow_it(self) -> list[str]:
        """Return assets not present in the known inventory."""
        raise NotImplementedError

    def what_would_they_hit_first(self) -> list[str]:
        """Return assets ranked by adversary priority."""
        raise NotImplementedError

    def brief(self) -> Brief:
        """Generate a narrative red team brief."""
        raise NotImplementedError

    def mitre_mapping(self) -> dict[str, list[str]]:
        """Return findings mapped to ATT&CK tactics and techniques."""
        raise NotImplementedError

    def remediation_plan(self) -> list[str]:
        """Return a prioritised fix list with effort estimates."""
        raise NotImplementedError

    def delta(self, previous: Report) -> Report:
        """Return a diff report showing what changed since *previous*."""
        raise NotImplementedError

    def audit_log(self) -> str:
        """Return a tamper-evident chain of custody for this engagement."""
        return self._audit.to_text()

    def assert_no_new_criticals(self) -> None:
        """Raise if new critical findings were discovered.

        Use as a CI/CD gate::

            report.assert_no_new_criticals()
        """
        raise NotImplementedError

    def export(self, format: str, path: str | None = None) -> str:  # noqa: A002
        """Export the report to *format*.

        Args:
            format: ``"sarif"``, ``"asff"``, ``"markdown"``, or ``"pdf"``.
            path: Optional file path to write the output to.

        Returns:
            The exported content as a string (SARIF JSON, Markdown, etc.).
        """
        raise NotImplementedError

class StreamEvent(BaseModel):
    """A single finding or status event emitted during ``stream=True`` mode.

    Attributes:
        level: Severity label - ``"CRITICAL"``, ``"HIGH"``, ``"MEDIUM"``,
            ``"LOW"``, or ``"INFO"``.
        description: Human-readable description of what was found or observed.
        asset: The asset that triggered this event (URL, IP, file path, etc.).
        _halt_fn: Internal callable to stop the engagement early.
    """

    level: str
    description: str
    asset: str = ""

    model_config = {"arbitrary_types_allowed": True}

    _halt_fn: Callable[[], None] | None = PrivateAttr(default=None)

    def halt(self) -> None:
        """Stop the running engagement immediately.

        Safe to call from inside a ``for event in m.engage(..., stream=True)``
        loop - the generator will stop after this event is yielded.
        """
        if self._halt_fn is not None:
            self._halt_fn()

class AuditEntry(BaseModel):
    """A single immutable record in the engagement audit trail.

    Attributes:
        timestamp: UTC timestamp of the event.
        action: Short action label (e.g. ``"scan_started"``, ``"plugin_run"``).
        actor: Who or what performed the action (plugin name, ``"llm"``, etc.).
        target: The asset or resource the action was performed against.
        detail: Free-form detail string for human review.
        checksum: SHA-256 of the entry fields for tamper detection.
    """

    timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC))
    action: str
    actor: str
    target: str
    detail: str = ""
    checksum: str = ""

    def model_post_init(self, __context: object) -> None:  # noqa: ANN001
        if not self.checksum:
            payload = json.dumps(
                {
                    "timestamp": self.timestamp.isoformat(),
                    "action": self.action,
                    "actor": self.actor,
                    "target": self.target,
                    "detail": self.detail,
                },
                sort_keys=True,
            )
            object.__setattr__(self, "checksum", hashlib.sha256(payload.encode()).hexdigest())

class AuditLog(BaseModel):
    """Ordered, tamper-evident chain of custody for an engagement.

    Attributes:
        engagement_id: Unique identifier for this engagement run.
        entries: Ordered list of audit entries.
    """

    engagement_id: str
    entries: list[AuditEntry] = []

    def append(self, entry: AuditEntry) -> None:
        """Append *entry* to the log."""
        self.entries.append(entry)

    def to_text(self) -> str:
        """Return a human-readable audit trail."""
        lines = [f"Engagement: {self.engagement_id}", ""]
        for e in self.entries:
            lines.append(f"[{e.timestamp.isoformat()}] {e.action} | {e.actor} → {e.target}")
            if e.detail:
                lines.append(f"  {e.detail}")
            lines.append(f"  checksum: {e.checksum}")
        return "\n".join(lines)