Implement KG-WP-0002 posture pilot scaffold

src/kings_guard/posture.py

@@ -0,0 +1,242 @@
+from __future__ import annotations
+
+from kings_guard.contracts import (
+    EffectorRequest,
+    ImmuneObservation,
+    ImmuneSignal,
+    PostureAssessment,
+    PostureEvaluation,
+    PostureLevel,
+    SecurityGenome,
+    SecurityPhenotype,
+    SignalKind,
+)
+
+CRITICAL_FINDINGS = frozenset(
+    {
+        "tenant_mismatch",
+        "undeclared_capability",
+        "unexpected_egress_destination",
+        "credential_exfil_probe",
+    }
+)
+ELEVATED_FINDINGS = frozenset(
+    {
+        "constraint_probe_or_boundary_trip",
+        "unexpected_protocol",
+        "unverified_identity_binding",
+        "control_plane_error",
+    }
+)
+RISK_WEIGHTS = {
+    "tenant_mismatch": 50,
+    "undeclared_capability": 40,
+    "unexpected_egress_destination": 50,
+    "credential_exfil_probe": 60,
+    "constraint_probe_or_boundary_trip": 25,
+    "unexpected_protocol": 15,
+    "unverified_identity_binding": 20,
+    "control_plane_error": 20,
+}
+
+
+class PostureEvaluator:
+    def evaluate(self, genome: SecurityGenome, observation: ImmuneObservation) -> PostureEvaluation:
+        phenotype = self._derive_phenotype(genome, observation)
+        assessment = self._assess(phenotype, observation)
+        signals = self._build_signals(observation, assessment)
+        return PostureEvaluation(phenotype=phenotype, assessment=assessment, signals=signals)
+
+    def _derive_phenotype(
+        self,
+        genome: SecurityGenome,
+        observation: ImmuneObservation,
+    ) -> SecurityPhenotype:
+        findings: list[str] = []
+        tolerated: list[str] = []
+
+        if observation.tenant_id != genome.tenant_id:
+            findings.append("tenant_mismatch")
+        if observation.capability not in genome.permitted_capabilities:
+            findings.append("undeclared_capability")
+        if observation.protocol not in genome.permitted_protocols:
+            findings.append("unexpected_protocol")
+        if (
+            observation.egress_destination is not None
+            and observation.egress_destination not in genome.permitted_egress
+        ):
+            findings.append("unexpected_egress_destination")
+
+        if observation.identity_binding and observation.identity_binding != "verified_token":
+            tolerance = _matching_tolerance(genome, "identity_binding", observation.identity_binding)
+            if tolerance is None:
+                findings.append("unverified_identity_binding")
+            else:
+                tolerated.append(f"tolerated:{tolerance.tolerance_id}")
+
+        if observation.decision.value == "deny" and observation.deny_reason == "credential_exfil":
+            findings.append("credential_exfil_probe")
+        elif observation.decision.value == "deny" and observation.deny_reason == "arg_constraint":
+            findings.append("constraint_probe_or_boundary_trip")
+        elif observation.decision.value == "error":
+            findings.append("control_plane_error")
+
+        return SecurityPhenotype(
+            subject_id=observation.subject_id,
+            tenant_id=observation.tenant_id,
+            observed_capability=observation.capability,
+            protocol=observation.protocol,
+            decision=observation.decision,
+            active_findings=tuple(findings),
+            tolerated_findings=tuple(tolerated),
+        )
+
+    def _assess(
+        self,
+        phenotype: SecurityPhenotype,
+        observation: ImmuneObservation,
+    ) -> PostureAssessment:
+        findings = set(phenotype.active_findings)
+
+        posture = PostureLevel.HEALTHY
+        if findings & CRITICAL_FINDINGS:
+            posture = PostureLevel.INFLAMED
+        elif findings:
+            posture = PostureLevel.ELEVATED
+        if observation.decision.value == "allow" and "unexpected_egress_destination" in findings:
+            posture = PostureLevel.COMPROMISED
+
+        risk_score = 5 + sum(RISK_WEIGHTS.get(item, 10) for item in findings)
+        if posture is PostureLevel.ELEVATED:
+            risk_score = max(risk_score, 40)
+        elif posture is PostureLevel.INFLAMED:
+            risk_score = max(risk_score, 80)
+        elif posture is PostureLevel.COMPROMISED:
+            risk_score = max(risk_score, 95)
+        risk_score = min(risk_score, 100)
+
+        confidence_score = 70
+        if observation.policy_version is not None:
+            confidence_score += 10
+        if observation.latency_ms is not None:
+            confidence_score += 5
+        if observation.resource_scope:
+            confidence_score += 5
+        confidence_score = min(confidence_score, 95)
+
+        rationale = _build_rationale(
+            posture=posture,
+            findings=phenotype.active_findings,
+            tolerated_findings=phenotype.tolerated_findings,
+        )
+        return PostureAssessment(
+            posture=posture,
+            risk_score=risk_score,
+            confidence_score=confidence_score,
+            findings=phenotype.active_findings,
+            tolerated_findings=phenotype.tolerated_findings,
+            rationale=rationale,
+        )
+
+    def _build_signals(
+        self,
+        observation: ImmuneObservation,
+        assessment: PostureAssessment,
+    ) -> tuple[ImmuneSignal, ...]:
+        if assessment.posture is PostureLevel.HEALTHY:
+            return ()
+
+        if observation.source_system == "qonto-assistant":
+            return (self._build_qonto_pilot_signal(observation, assessment),)
+
+        signal = ImmuneSignal(
+            signal_id=f"sig:{observation.observation_id}",
+            signal_kind=SignalKind.OBSERVATION_ALERT,
+            posture=assessment.posture,
+            summary=assessment.rationale,
+            target_system=observation.source_system,
+            findings=assessment.findings,
+            metadata={"source_system": observation.source_system},
+        )
+        return (signal,)
+
+    def _build_qonto_pilot_signal(
+        self,
+        observation: ImmuneObservation,
+        assessment: PostureAssessment,
+    ) -> ImmuneSignal:
+        if "credential_exfil_probe" in assessment.findings:
+            action = "lock_actor_temporarily"
+            reason = (
+                "Observed a credential-exfil deny signal; qonto-assistant should activate "
+                "its fast local loop lockout and preserve metadata-only evidence."
+            )
+        else:
+            action = "tighten_actor_scrutiny"
+            reason = (
+                "Observed repeated policy-boundary pressure; qonto-assistant should tighten "
+                "local scrutiny without delegating final authorization to kings-guard."
+            )
+
+        return ImmuneSignal(
+            signal_id=f"sig:{observation.observation_id}",
+            signal_kind=SignalKind.POSTURE_HINT,
+            posture=assessment.posture,
+            summary=reason,
+            target_system="qonto-assistant",
+            findings=assessment.findings,
+            effector_requests=(
+                EffectorRequest(
+                    target_system="qonto-assistant",
+                    action=action,
+                    authority_boundary="advisory_only",
+                    reason=reason,
+                    requires_human_approval=False,
+                ),
+                EffectorRequest(
+                    target_system="state-hub",
+                    action="record_non_secret_incident_evidence",
+                    authority_boundary="metadata_only",
+                    reason="Preserve posture evidence without copying secret values.",
+                    requires_human_approval=False,
+                ),
+            ),
+            metadata={
+                "pilot_lane": "qonto-assistant",
+                "resource_scope": observation.resource_scope or "unknown",
+            },
+        )
+
+
+def _matching_tolerance(
+    genome: SecurityGenome,
+    field_name: str,
+    field_value: str,
+):
+    for tolerance in genome.tolerances:
+        if tolerance.match_field == field_name and tolerance.match_value == field_value:
+            return tolerance
+    return None
+
+
+def _build_rationale(
+    *,
+    posture: PostureLevel,
+    findings: tuple[str, ...],
+    tolerated_findings: tuple[str, ...],
+) -> str:
+    if posture is PostureLevel.HEALTHY:
+        if tolerated_findings:
+            return (
+                "Healthy posture with tolerated deviations only: "
+                + ", ".join(tolerated_findings)
+            )
+        return "Healthy posture: observation is compatible with declared intent."
+
+    detail = ", ".join(findings) if findings else "no active findings"
+    tolerated = (
+        f" Tolerated deviations still present: {', '.join(tolerated_findings)}."
+        if tolerated_findings
+        else ""
+    )
+    return f"{posture.value.title()} posture driven by {detail}.{tolerated}"