242 lines
8.5 KiB
Python
242 lines
8.5 KiB
Python
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}"
|