Skip to content

API Reference: Safety

This section provides technical documentation for the safety modules, automatically generated from the source code.

Analyzer

safety.analyzer

Classes

Finding

Bases: BaseModel

Cryptographically signed finding package.

Source code in src/safety/analyzer.py 
29
30
31
32
33
34
35
36
37
38
class Finding(BaseModel):
    """Cryptographically signed finding package."""
    agent_id: str
    incident_id: str
    incident_type: str
    severity: str
    proposed_remediation: Dict[str, Any]
    timestamp: int  # Integer timestamp for cryptographic stability
    nonce: str
    metadata: Dict[str, Any] = {} # For armor safety scores

ModelArmor

Simulates GCP Model Armor / Safety Guardrails. Sanitizes agent output before it reaches the consensus layer.

Source code in src/safety/analyzer.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
class ModelArmor:
    """
    Simulates GCP Model Armor / Safety Guardrails.
    Sanitizes agent output before it reaches the consensus layer.
    """
    # Pattern for potential GCP/API keys
    SECRET_PATTERN = re.compile(r"(AIza[0-9A-Za-z-_]{35}|sk-[a-zA-Z0-9]{48})")

    def sanitize_finding(self, finding: Finding) -> Finding:
        """Surgically redacts secrets and adds safety metadata."""
        finding_json = finding.model_dump_json()

        # 1. Leak Detection (Simple Regex-based scanning)
        if self.SECRET_PATTERN.search(finding_json):
            log_event(logger, logging.WARNING, "ModelArmor: Secret pattern detected. Redacting.", extra={
                "agent_id": finding.agent_id,
                "incident_id": finding.incident_id,
                "armor_action": "REDACT"
            })
            # Redact common fields
            if "description" in finding.proposed_remediation:
                finding.proposed_remediation["description"] = self.SECRET_PATTERN.sub("[REDACTED_SECRET]", finding.proposed_remediation["description"])

        # 2. Add safety metadata
        finding.metadata["safety_score"] = 0.99
        finding.metadata["armor_status"] = "VERIFIED_CLEAN"
        return finding
Functions
sanitize_finding(finding)

Surgically redacts secrets and adds safety metadata.

Source code in src/safety/analyzer.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def sanitize_finding(self, finding: Finding) -> Finding:
    """Surgically redacts secrets and adds safety metadata."""
    finding_json = finding.model_dump_json()

    # 1. Leak Detection (Simple Regex-based scanning)
    if self.SECRET_PATTERN.search(finding_json):
        log_event(logger, logging.WARNING, "ModelArmor: Secret pattern detected. Redacting.", extra={
            "agent_id": finding.agent_id,
            "incident_id": finding.incident_id,
            "armor_action": "REDACT"
        })
        # Redact common fields
        if "description" in finding.proposed_remediation:
            finding.proposed_remediation["description"] = self.SECRET_PATTERN.sub("[REDACTED_SECRET]", finding.proposed_remediation["description"])

    # 2. Add safety metadata
    finding.metadata["safety_score"] = 0.99
    finding.metadata["armor_status"] = "VERIFIED_CLEAN"
    return finding

VertexAIAnalyzer

Hardened Agentic Analyzer using Vertex AI (Gemini) for incident root-cause analysis. Supports both research simulation and production GCP modes with exponential backoff.

Source code in src/safety/analyzer.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
class VertexAIAnalyzer:
    """
    Hardened Agentic Analyzer using Vertex AI (Gemini) for incident root-cause analysis.
    Supports both research simulation and production GCP modes with exponential backoff.
    """
    def __init__(self, agent_id: str, private_key: rsa.RSAPrivateKey, project_id: str = None, location: str = "us-central1"):
        self.agent_id = agent_id
        self._private_key = private_key
        self.project_id = project_id or os.getenv("GCP_PROJECT_ID")
        self.location = location
        self._initialized_gcp = False
        self.armor = ModelArmor()

    def _init_gcp(self):
        if not HAS_GCP:
            raise ImportError("google-cloud-aiplatform or vertexai not installed. Run 'pip install .[gcp]'")
        if not self._initialized_gcp:
            # Supports ADC or explicit GOOGLE_APPLICATION_CREDENTIALS env var
            vertexai.init(project=self.project_id, location=self.location)
            self._initialized_gcp = True
            logger.info(f"Vertex AI initialized for project {self.project_id}")


    def analyze_logs(self, logs: List[Dict[str, Any]], mode: str = "simulate") -> List[Finding]:
        """
        Analyzes logs to detect incidents and propose remediation.
        """
        if mode == "real":
            return self._analyze_real_with_retry(logs)

        # Simulation Logic (Research Mode)
        findings = []
        for log in logs:
            msg = str(log.get("jsonPayload", {}).get("message", "")).lower()
            if "oom" in msg or "critical" in msg:
                finding = Finding(
                    agent_id=self.agent_id,
                    incident_id=hashlib.sha256(msg.encode()).hexdigest()[:8],
                    incident_type="oomkill",
                    severity="CRITICAL",
                    proposed_remediation={
                        "operation": "SCALE_UP",
                        "replicas": 2,
                        "target": "api-service",
                        "description": "Detected critical OOM signal in logs."
                    },
                    timestamp=int(time.time()),
                    nonce=os.urandom(16).hex()
                )

                findings.append(self.armor.sanitize_finding(finding))
        return findings

    def fetch_and_analyze(self, query: str, log_source: LogSourcePort, limit: int = 100, mode: str = "simulate") -> List[Finding]:
        """Fetches logs using the injected LogSourcePort and processes them."""
        logs = log_source.fetch_recent_logs(query, limit)
        return self.analyze_logs(logs, mode)

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        reraise=True
    )
    def _analyze_real_with_retry(self, logs: List[Dict[str, Any]]) -> List[Finding]:
        """Wrapper for real mode with production-reference retries."""
        return self._analyze_real(logs)

    def _analyze_real(self, logs: List[Dict[str, Any]]) -> List[Finding]:
        """Performs real Vertex AI analysis using Gemini 1.5 Pro."""
        self._init_gcp()
        model = GenerativeModel("gemini-1.5-pro")

        # Enterprise System Prompt for structured output
        system_instr = "You are a SRE Automation Agent. Analyze logs and return a valid JSON Finding array."
        prompt = f"{system_instr}\n\nLogs to analyze: {json.dumps(logs)}"

        try:
            # Exhaustive safety configuration
            safety_settings = [
                SafetySetting(category=HarmCategory.HARM_CATEGORY_HATE_SPEECH, threshold=HarmBlockThreshold.BLOCK_ONLY_HIGH),
                SafetySetting(category=HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT, threshold=HarmBlockThreshold.BLOCK_ONLY_HIGH),
                SafetySetting(category=HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT, threshold=HarmBlockThreshold.BLOCK_ONLY_HIGH),
                SafetySetting(category=HarmCategory.HARM_CATEGORY_HARASSMENT, threshold=HarmBlockThreshold.BLOCK_ONLY_HIGH),
            ]

            response = model.generate_content(
                prompt,
                generation_config={"response_mime_type": "application/json"},
                safety_settings=safety_settings
            )

            # Deterministic JSON Mapping
            raw_text = response.text
            remediation_data = json.loads(raw_text)

            # Map Gemini output to our Finding model
            findings = []
            if isinstance(remediation_data, list):
                for item in remediation_data:
                    finding = Finding(
                        agent_id=self.agent_id,
                        incident_id=item.get("id", f"gemini-{os.urandom(4).hex()}"),
                        incident_type=item.get("type", "automated-detection"),
                        severity=item.get("severity", "MEDIUM"),
                        proposed_remediation=item.get("remediation", {"operation": "NOTIFY", "target": "sre-oncall"}),
                        timestamp=int(time.time()),
                        nonce=os.urandom(16).hex()
                    )

                    # Advisory Authenticity Scoring (Phase 2)
                    if HAS_INTELLIGENCE:
                        auth_result = self.authenticity_scorer.score_proposal(finding.proposed_remediation.get("description", ""), None)
                        finding.metadata.update(auth_result)

                    findings.append(self.armor.sanitize_finding(finding))

                log_event(logger, logging.INFO, "Vertex AI generated findings.", extra={
                    "agent_id": self.agent_id,
                    "incident_count": len(findings),
                    "mode": "real"
                })
            return findings

        except json.JSONDecodeError:
            logger.error("Gemini returned invalid JSON format.")
            raise ValueError("Failed to parse AI output into valid Finding model.")
        except Exception as e:
            logger.error(f"Vertex AI API call failed: {e}")
            raise

    def sign_finding(self, finding: Finding) -> Dict[str, Any]:
        """Signs the finding using the agent's private RSA key."""
        # Use canonical JSON (sorted keys, no whitespace) for stable hashing
        finding_json = json.dumps(finding.model_dump(), sort_keys=True, separators=(',', ':'))
        logger.debug(f"Signing proposal JSON: {finding_json}")
        signature = self._private_key.sign(
            finding_json.encode(),
            padding.PSS(
                mgf=padding.MGF1(hashes.SHA256()),
                salt_length=padding.PSS.MAX_LENGTH
            ),
            hashes.SHA256()
        )
        log_event(logger, logging.INFO, "Finding signed successfully.", extra={
            "agent_id": self.agent_id,
            "incident_id": finding.incident_id,
            "nonce": finding.nonce,
            "timestamp": finding.timestamp
        })
        return {
            "agent_id": self.agent_id,
            "signature_hex": signature.hex(),
            "finding": finding.model_dump(),
            "timestamp": finding.timestamp,
            "nonce": finding.nonce
        }
Functions
analyze_logs(logs, mode='simulate')

Analyzes logs to detect incidents and propose remediation.

Source code in src/safety/analyzer.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def analyze_logs(self, logs: List[Dict[str, Any]], mode: str = "simulate") -> List[Finding]:
    """
    Analyzes logs to detect incidents and propose remediation.
    """
    if mode == "real":
        return self._analyze_real_with_retry(logs)

    # Simulation Logic (Research Mode)
    findings = []
    for log in logs:
        msg = str(log.get("jsonPayload", {}).get("message", "")).lower()
        if "oom" in msg or "critical" in msg:
            finding = Finding(
                agent_id=self.agent_id,
                incident_id=hashlib.sha256(msg.encode()).hexdigest()[:8],
                incident_type="oomkill",
                severity="CRITICAL",
                proposed_remediation={
                    "operation": "SCALE_UP",
                    "replicas": 2,
                    "target": "api-service",
                    "description": "Detected critical OOM signal in logs."
                },
                timestamp=int(time.time()),
                nonce=os.urandom(16).hex()
            )

            findings.append(self.armor.sanitize_finding(finding))
    return findings
fetch_and_analyze(query, log_source, limit=100, mode='simulate')

Fetches logs using the injected LogSourcePort and processes them.

Source code in src/safety/analyzer.py 
121
122
123
124
def fetch_and_analyze(self, query: str, log_source: LogSourcePort, limit: int = 100, mode: str = "simulate") -> List[Finding]:
    """Fetches logs using the injected LogSourcePort and processes them."""
    logs = log_source.fetch_recent_logs(query, limit)
    return self.analyze_logs(logs, mode)
sign_finding(finding)

Signs the finding using the agent's private RSA key.

Source code in src/safety/analyzer.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def sign_finding(self, finding: Finding) -> Dict[str, Any]:
    """Signs the finding using the agent's private RSA key."""
    # Use canonical JSON (sorted keys, no whitespace) for stable hashing
    finding_json = json.dumps(finding.model_dump(), sort_keys=True, separators=(',', ':'))
    logger.debug(f"Signing proposal JSON: {finding_json}")
    signature = self._private_key.sign(
        finding_json.encode(),
        padding.PSS(
            mgf=padding.MGF1(hashes.SHA256()),
            salt_length=padding.PSS.MAX_LENGTH
        ),
        hashes.SHA256()
    )
    log_event(logger, logging.INFO, "Finding signed successfully.", extra={
        "agent_id": self.agent_id,
        "incident_id": finding.incident_id,
        "nonce": finding.nonce,
        "timestamp": finding.timestamp
    })
    return {
        "agent_id": self.agent_id,
        "signature_hex": signature.hex(),
        "finding": finding.model_dump(),
        "timestamp": finding.timestamp,
        "nonce": finding.nonce
    }

Functions

Voting

safety.voting

Classes

ValidationError

Bases: Exception

Custom exception with context for validation failures.

Source code in src/safety/voting.py 
16
17
18
19
20
21
class ValidationError(Exception):
    """Custom exception with context for validation failures."""
    def __init__(self, message: str, code: str, details: Dict = None):
        self.code = code
        self.details = details or {}
        super().__init__(message)

VotingValidator

Validates agent signatures against a registered public key list to verify a quorum threshold has been reached.

Source code in src/safety/voting.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
class VotingValidator:
    """
    Validates agent signatures against a registered public key list
    to verify a quorum threshold has been reached.
    """
    def __init__(self, threshold: float = 0.66, max_clock_skew: int = 60):
        self.threshold = threshold
        self.max_clock_skew = max_clock_skew
        self._authorized_keys: Dict[str, rsa.RSAPublicKey] = {}
        self._seen_nonces = TTLCache(maxsize=10000, ttl=300)

    def register_agent(self, agent_id: str, public_key_pem: bytes):
        """Registers an authorized agent with its public key."""
        try:
            self._authorized_keys[agent_id] = serialization.load_pem_public_key(public_key_pem)
            logger.info(f"Registered agent: {agent_id}")
        except Exception as e:
            logger.error(f"Failed to register agent {agent_id}: {e}")
            raise ValidationError(f"Invalid public key for agent {agent_id}", "AUTH_CONFIG_ERROR")

    def verify_quorum(self, proposal: Dict[str, Any], signatures: List[AgentSignature]) -> VotingProof:
        """
        Verifies that a majority of registered agents signed the proposal.
        """
        # Use canonical JSON (sorted keys, no whitespace) for stable hashing
        proposal_json = json.dumps(proposal, sort_keys=True, separators=(',', ':'))
        decision_hash = hashlib.sha256(proposal_json.encode()).hexdigest()
        logger.debug(f"Hashed proposal: {decision_hash} | Content: {proposal_json[:50]}...")

        valid_signatures = []
        seen_agents = set()
        current_time = time.time()

        # Snapshot of seen nonces at start of verification run to allow same-nonce signatures in same batch
        initially_seen = set(self._seen_nonces.keys())

        for sig in signatures:
            try:
                # 1. Nonce check
                if sig.nonce in initially_seen:
                    raise ValidationError("Replay attack detected", "REPLAY_ATTACK", {"nonce": sig.nonce, "agent": sig.agent_id})

                # 2. Clock skew check
                skew = current_time - sig.timestamp
                if abs(skew) > self.max_clock_skew:
                    raise ValidationError("Proposal timestamp invalid", "STALE_PROPOSAL", {"timestamp": sig.timestamp, "skew": skew})

                # 3. Authorization check
                if sig.agent_id not in self._authorized_keys:
                    raise ValidationError("Unauthorized agent", "UNAUTHORIZED_AGENT", {"agent": sig.agent_id})

                # 4. Duplicate agent check
                if sig.agent_id in seen_agents:
                    raise ValidationError("Duplicate agent signature", "DUPLICATE_SIGNATURE", {"agent": sig.agent_id})

                # 5. Cryptographic Verification
                public_key = self._authorized_keys[sig.agent_id]
                public_key.verify(
                    bytes.fromhex(sig.signature_hex),
                    proposal_json.encode(),
                    padding.PSS(
                        mgf=padding.MGF1(hashes.SHA256()),
                        salt_length=padding.PSS.MAX_LENGTH
                    ),
                    hashes.SHA256()
                )

                valid_signatures.append(sig)
                seen_agents.add(sig.agent_id)
                self._seen_nonces[sig.nonce] = True

            except InvalidSignature:
                log_event(logger, logging.ERROR, "Invalid crypto signature.", extra={
                    "agent_id": sig.agent_id,
                    "decision_hash": decision_hash
                })
            except ValidationError as e:
                log_event(logger, logging.WARNING, f"Validation check failed: {str(e)}", extra={
                    "code": e.code,
                    **e.details
                })
            except Exception as e:
                log_event(logger, logging.ERROR, f"Unexpected error: {e}", extra={
                    "agent_id": sig.agent_id
                })

        total_agents = max(len(self._authorized_keys), 1)
        quorum_reached = len(valid_signatures) / total_agents >= self.threshold

        log_event(logger, logging.INFO, "Voting quorum evaluation complete.", extra={
            "quorum_reached": quorum_reached,
            "ratio": len(valid_signatures) / total_agents,
            "valid_count": len(valid_signatures),
            "total_agents": total_agents
        })

        return VotingProof(
            decision_hash=decision_hash,
            signatures=valid_signatures,
            quorum_reached=quorum_reached,
            threshold_used=self.threshold,
            total_authorized=len(self._authorized_keys)
        )
Functions
register_agent(agent_id, public_key_pem)

Registers an authorized agent with its public key.

Source code in src/safety/voting.py 
47
48
49
50
51
52
53
54
def register_agent(self, agent_id: str, public_key_pem: bytes):
    """Registers an authorized agent with its public key."""
    try:
        self._authorized_keys[agent_id] = serialization.load_pem_public_key(public_key_pem)
        logger.info(f"Registered agent: {agent_id}")
    except Exception as e:
        logger.error(f"Failed to register agent {agent_id}: {e}")
        raise ValidationError(f"Invalid public key for agent {agent_id}", "AUTH_CONFIG_ERROR")
verify_quorum(proposal, signatures)

Verifies that a majority of registered agents signed the proposal.

Source code in src/safety/voting.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def verify_quorum(self, proposal: Dict[str, Any], signatures: List[AgentSignature]) -> VotingProof:
    """
    Verifies that a majority of registered agents signed the proposal.
    """
    # Use canonical JSON (sorted keys, no whitespace) for stable hashing
    proposal_json = json.dumps(proposal, sort_keys=True, separators=(',', ':'))
    decision_hash = hashlib.sha256(proposal_json.encode()).hexdigest()
    logger.debug(f"Hashed proposal: {decision_hash} | Content: {proposal_json[:50]}...")

    valid_signatures = []
    seen_agents = set()
    current_time = time.time()

    # Snapshot of seen nonces at start of verification run to allow same-nonce signatures in same batch
    initially_seen = set(self._seen_nonces.keys())

    for sig in signatures:
        try:
            # 1. Nonce check
            if sig.nonce in initially_seen:
                raise ValidationError("Replay attack detected", "REPLAY_ATTACK", {"nonce": sig.nonce, "agent": sig.agent_id})

            # 2. Clock skew check
            skew = current_time - sig.timestamp
            if abs(skew) > self.max_clock_skew:
                raise ValidationError("Proposal timestamp invalid", "STALE_PROPOSAL", {"timestamp": sig.timestamp, "skew": skew})

            # 3. Authorization check
            if sig.agent_id not in self._authorized_keys:
                raise ValidationError("Unauthorized agent", "UNAUTHORIZED_AGENT", {"agent": sig.agent_id})

            # 4. Duplicate agent check
            if sig.agent_id in seen_agents:
                raise ValidationError("Duplicate agent signature", "DUPLICATE_SIGNATURE", {"agent": sig.agent_id})

            # 5. Cryptographic Verification
            public_key = self._authorized_keys[sig.agent_id]
            public_key.verify(
                bytes.fromhex(sig.signature_hex),
                proposal_json.encode(),
                padding.PSS(
                    mgf=padding.MGF1(hashes.SHA256()),
                    salt_length=padding.PSS.MAX_LENGTH
                ),
                hashes.SHA256()
            )

            valid_signatures.append(sig)
            seen_agents.add(sig.agent_id)
            self._seen_nonces[sig.nonce] = True

        except InvalidSignature:
            log_event(logger, logging.ERROR, "Invalid crypto signature.", extra={
                "agent_id": sig.agent_id,
                "decision_hash": decision_hash
            })
        except ValidationError as e:
            log_event(logger, logging.WARNING, f"Validation check failed: {str(e)}", extra={
                "code": e.code,
                **e.details
            })
        except Exception as e:
            log_event(logger, logging.ERROR, f"Unexpected error: {e}", extra={
                "agent_id": sig.agent_id
            })

    total_agents = max(len(self._authorized_keys), 1)
    quorum_reached = len(valid_signatures) / total_agents >= self.threshold

    log_event(logger, logging.INFO, "Voting quorum evaluation complete.", extra={
        "quorum_reached": quorum_reached,
        "ratio": len(valid_signatures) / total_agents,
        "valid_count": len(valid_signatures),
        "total_agents": total_agents
    })

    return VotingProof(
        decision_hash=decision_hash,
        signatures=valid_signatures,
        quorum_reached=quorum_reached,
        threshold_used=self.threshold,
        total_authorized=len(self._authorized_keys)
    )

Functions

Safety Gate

safety.safety_gate

Classes

ActionProposal

Bases: BaseModel

Schema for a remediation proposal.

Source code in src/safety/safety_gate.py 
 9
10
11
12
13
14
15
class ActionProposal(BaseModel):
    """Schema for a remediation proposal."""
    operation: str
    target: str
    replicas: Optional[int] = 0
    scale_factor: Optional[float] = 1.0
    current_state: Optional[Dict[str, Any]] = {}

GateResult

Bases: BaseModel

Rich result object for safety evaluations.

Source code in src/safety/safety_gate.py 
31
32
33
34
35
36
37
class GateResult(BaseModel):
    """Rich result object for safety evaluations."""
    allowed: bool
    risk_score: float = 0.0
    estimated_cost_increase: float = 0.0
    reason: str = ""
    blocked_operation: Optional[str] = None

SafetyConfig

Bases: BaseSettings

Deterministic safety boundaries. Loads from environment variables with SAFETY_ prefix.

Source code in src/safety/safety_gate.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
class SafetyConfig(BaseSettings):
    """
    Deterministic safety boundaries.
    Loads from environment variables with SAFETY_ prefix.
    """
    max_replicas_per_service: int = 20
    max_scale_factor: float = 2.0
    max_estimated_cost_per_remediation: float = 50.0
    allowed_operations: List[str] = ["SCALE_UP", "RESTART", "NOTIFY", "UPDATE"]
    cost_per_replica_hour: float = 0.05
    authorized_agents: List[str] = []

    model_config = SettingsConfigDict(env_prefix='SAFETY_', env_file='.env')

SafetyGate

Deterministic Safety Validation Gate. Enforces resource quotas, cost limits, and operational restrictions.

Source code in src/safety/safety_gate.py 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class SafetyGate:
    """
    Deterministic Safety Validation Gate.
    Enforces resource quotas, cost limits, and operational restrictions.
    """
    def __init__(self, config: SafetyConfig):
        self.config = config

    def evaluate(self, raw_proposal: Dict[str, Any]) -> GateResult:
        """
        Validates a proposal against resource, operational, and cost boundaries.
        """
        try:
            proposal = ActionProposal(**raw_proposal)
        except Exception as e:
            logger.error(f"Invalid proposal schema: {e}")
            return GateResult(
                allowed=False,
                reason=f"Proposal failed schema validation: {e}",
                risk_score=1.0
            )

        # 1. Enforce Allow-List
        op = proposal.operation.upper()
        if op not in self.config.allowed_operations:
            log_event(logger, logging.WARNING, "Operation NOT in allow-list.", extra={
                "operation": op,
                "allowed": self.config.allowed_operations,
                "action": "BLOCK"
            })
            return GateResult(
                allowed=False, 
                reason=f"Operation '{op}' is not in the approved safety allow-list.",
                risk_score=0.9,
                blocked_operation=op
            )

        # 2. Check scale factor
        if proposal.scale_factor > self.config.max_scale_factor:
            logger.warning(f"Excessive scale factor: {proposal.scale_factor}")
            return GateResult(
                allowed=False,
                reason=f"Scale factor {proposal.scale_factor} exceeds safety limit.",
                risk_score=0.8
            )

        # 3. Check replica scaling
        if proposal.replicas > self.config.max_replicas_per_service:
            logger.warning(f"Excessive replicas: {proposal.replicas}")
            return GateResult(
                allowed=False,
                reason=f"Replica count {proposal.replicas} exceeds safety limit.",
                risk_score=0.85
            )

        # 4. Cost Guard
        estimated_cost = proposal.replicas * self.config.cost_per_replica_hour
        if estimated_cost > self.config.max_estimated_cost_per_remediation:
            logger.warning(f"Cost guard blocked remediation: ${estimated_cost:.2f}")
            return GateResult(
                allowed=False,
                reason=f"Estimated cost ${estimated_cost:.2f} exceeds threshold.",
                risk_score=0.7,
                estimated_cost_increase=estimated_cost
            )

        log_event(logger, logging.INFO, "Safety Gate: APPROVED.", extra={
            "target": proposal.target,
            "operation": op,
            "estimated_cost": estimated_cost,
            "risk_score": 0.1
        })
        return GateResult(
            allowed=True,
            reason="All gates passed.",
            estimated_cost_increase=estimated_cost,
            risk_score=0.1
        )
Functions
evaluate(raw_proposal)

Validates a proposal against resource, operational, and cost boundaries.

Source code in src/safety/safety_gate.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def evaluate(self, raw_proposal: Dict[str, Any]) -> GateResult:
    """
    Validates a proposal against resource, operational, and cost boundaries.
    """
    try:
        proposal = ActionProposal(**raw_proposal)
    except Exception as e:
        logger.error(f"Invalid proposal schema: {e}")
        return GateResult(
            allowed=False,
            reason=f"Proposal failed schema validation: {e}",
            risk_score=1.0
        )

    # 1. Enforce Allow-List
    op = proposal.operation.upper()
    if op not in self.config.allowed_operations:
        log_event(logger, logging.WARNING, "Operation NOT in allow-list.", extra={
            "operation": op,
            "allowed": self.config.allowed_operations,
            "action": "BLOCK"
        })
        return GateResult(
            allowed=False, 
            reason=f"Operation '{op}' is not in the approved safety allow-list.",
            risk_score=0.9,
            blocked_operation=op
        )

    # 2. Check scale factor
    if proposal.scale_factor > self.config.max_scale_factor:
        logger.warning(f"Excessive scale factor: {proposal.scale_factor}")
        return GateResult(
            allowed=False,
            reason=f"Scale factor {proposal.scale_factor} exceeds safety limit.",
            risk_score=0.8
        )

    # 3. Check replica scaling
    if proposal.replicas > self.config.max_replicas_per_service:
        logger.warning(f"Excessive replicas: {proposal.replicas}")
        return GateResult(
            allowed=False,
            reason=f"Replica count {proposal.replicas} exceeds safety limit.",
            risk_score=0.85
        )

    # 4. Cost Guard
    estimated_cost = proposal.replicas * self.config.cost_per_replica_hour
    if estimated_cost > self.config.max_estimated_cost_per_remediation:
        logger.warning(f"Cost guard blocked remediation: ${estimated_cost:.2f}")
        return GateResult(
            allowed=False,
            reason=f"Estimated cost ${estimated_cost:.2f} exceeds threshold.",
            risk_score=0.7,
            estimated_cost_increase=estimated_cost
        )

    log_event(logger, logging.INFO, "Safety Gate: APPROVED.", extra={
        "target": proposal.target,
        "operation": op,
        "estimated_cost": estimated_cost,
        "risk_score": 0.1
    })
    return GateResult(
        allowed=True,
        reason="All gates passed.",
        estimated_cost_increase=estimated_cost,
        risk_score=0.1
    )

Functions

Remediator

safety.remediator

Classes

DryRunRemediator

Bases: ActuationPort

Verified actuator that checks voting quorum and safety gates before 'execution'.

Source code in src/safety/remediator.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class DryRunRemediator(ActuationPort):
    """
    Verified actuator that checks voting quorum and safety gates before 'execution'.
    """
    def __init__(self, voting: VotingValidator, safety_gate: SafetyGate):
        self.voting = voting
        self.safety_gate = safety_gate

    def verify_remediation_signatures(self, finding: Dict[str, Any], agent_sigs: List[AgentSignature]) -> bool:
        """
        Enforces cryptographic signature validation.
        Fails closed if the payload has not achieved quorum verification.
        """
        try:
            voting_proof = self.voting.verify_quorum(finding, agent_sigs)
            return voting_proof.quorum_reached
        except Exception as e:
            logger.error(f"Signature verification threw exception: {e}")
            return False

    def apply_patch(self, target: str, operation: str, params: Dict[str, Any]) -> bool:
        """Implements the ActuationPort interface."""
        action_msg = f"ActuationPort: Executed {operation} on {target}"
        logger.info(action_msg)
        return True

    def process_proposal(self, finding: Dict[str, Any], signatures: List[Dict[str, Any]]) -> RemediationResult:
        """
        Verifies and processes a remediation proposal based on agent quorum.
        """
        # 1. Verify signatures format
        try:
            agent_sigs = [AgentSignature(**s) for s in signatures]
        except Exception as e:
            logger.error(f"Invalid signature format: {e}")
            return RemediationResult(
                success=False,
                action_taken="NONE",
                message=f"Invalid signature format: {e}",
                safety_check_passed=False,
                consensus_check_passed=False
            )

        # 2. Cryptographic signature check (Fail closed)
        consensus_passed = self.verify_remediation_signatures(finding, agent_sigs)
        if not consensus_passed:
            logger.warning("Voting quorum validation FAILED. Rejecting state-changing action.")
            return RemediationResult(
                success=False,
                action_taken="NONE",
                message="Quorum verification failed.",
                safety_check_passed=False,
                consensus_check_passed=False
            )

        # 3. Verify Safety Gate boundaries
        remediation = finding.get("proposed_remediation", {})
        if "target" not in remediation:
            remediation["target"] = finding.get("incident_id", "unknown-target")

        gate_result = self.safety_gate.evaluate(remediation)

        if not gate_result.allowed:
            logger.warning(f"Safety gate BLOCKED: {gate_result.reason}")
            return RemediationResult(
                success=False,
                action_taken="NONE",
                message=f"Safety gate blocked operation: {gate_result.reason}",
                safety_check_passed=False,
                consensus_check_passed=True
            )

        # 4. Execute Actuation using ports abstract boundary
        operation = remediation.get("operation", "UNKNOWN")
        target = remediation.get("target", "unknown-target")
        actuation_success = self.apply_patch(target, operation, remediation)

        return RemediationResult(
            success=actuation_success,
            action_taken=operation,
            message=f"Executed {operation} on {target} successfully.",
            safety_check_passed=True,
            consensus_check_passed=True
        )
Functions
apply_patch(target, operation, params)

Implements the ActuationPort interface.

Source code in src/safety/remediator.py 
37
38
39
40
41
def apply_patch(self, target: str, operation: str, params: Dict[str, Any]) -> bool:
    """Implements the ActuationPort interface."""
    action_msg = f"ActuationPort: Executed {operation} on {target}"
    logger.info(action_msg)
    return True
process_proposal(finding, signatures)

Verifies and processes a remediation proposal based on agent quorum.

Source code in src/safety/remediator.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def process_proposal(self, finding: Dict[str, Any], signatures: List[Dict[str, Any]]) -> RemediationResult:
    """
    Verifies and processes a remediation proposal based on agent quorum.
    """
    # 1. Verify signatures format
    try:
        agent_sigs = [AgentSignature(**s) for s in signatures]
    except Exception as e:
        logger.error(f"Invalid signature format: {e}")
        return RemediationResult(
            success=False,
            action_taken="NONE",
            message=f"Invalid signature format: {e}",
            safety_check_passed=False,
            consensus_check_passed=False
        )

    # 2. Cryptographic signature check (Fail closed)
    consensus_passed = self.verify_remediation_signatures(finding, agent_sigs)
    if not consensus_passed:
        logger.warning("Voting quorum validation FAILED. Rejecting state-changing action.")
        return RemediationResult(
            success=False,
            action_taken="NONE",
            message="Quorum verification failed.",
            safety_check_passed=False,
            consensus_check_passed=False
        )

    # 3. Verify Safety Gate boundaries
    remediation = finding.get("proposed_remediation", {})
    if "target" not in remediation:
        remediation["target"] = finding.get("incident_id", "unknown-target")

    gate_result = self.safety_gate.evaluate(remediation)

    if not gate_result.allowed:
        logger.warning(f"Safety gate BLOCKED: {gate_result.reason}")
        return RemediationResult(
            success=False,
            action_taken="NONE",
            message=f"Safety gate blocked operation: {gate_result.reason}",
            safety_check_passed=False,
            consensus_check_passed=True
        )

    # 4. Execute Actuation using ports abstract boundary
    operation = remediation.get("operation", "UNKNOWN")
    target = remediation.get("target", "unknown-target")
    actuation_success = self.apply_patch(target, operation, remediation)

    return RemediationResult(
        success=actuation_success,
        action_taken=operation,
        message=f"Executed {operation} on {target} successfully.",
        safety_check_passed=True,
        consensus_check_passed=True
    )
verify_remediation_signatures(finding, agent_sigs)

Enforces cryptographic signature validation. Fails closed if the payload has not achieved quorum verification.

Source code in src/safety/remediator.py 
25
26
27
28
29
30
31
32
33
34
35
def verify_remediation_signatures(self, finding: Dict[str, Any], agent_sigs: List[AgentSignature]) -> bool:
    """
    Enforces cryptographic signature validation.
    Fails closed if the payload has not achieved quorum verification.
    """
    try:
        voting_proof = self.voting.verify_quorum(finding, agent_sigs)
        return voting_proof.quorum_reached
    except Exception as e:
        logger.error(f"Signature verification threw exception: {e}")
        return False

Functions

Runtime Security

safety.security

Classes

RuntimeSecurity

Verifies cryptographic identity of agents.

Source code in src/safety/security.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class RuntimeSecurity:
    """
    Verifies cryptographic identity of agents.
    """
    def __init__(self, platform_public_key_pem: bytes):
        self.public_key = serialization.load_pem_public_key(platform_public_key_pem)

    def verify_runtime_attestation(self, attestation_report: bytes, signature: bytes) -> bool:
        """
        Verifies that the runtime attestation report was signed by the 
        trusted platform (e.g. simulated TPM or Cloud HSM).
        NOTE: In production, the public key should be fetched from GCP Secret Manager 
        or a KMS-backed Identity Provider with automated rotation enabled.
        """
        try:
            self.public_key.verify(
                signature,
                attestation_report,
                padding.PSS(
                    mgf=padding.MGF1(hashes.SHA256()),
                    salt_length=padding.PSS.MAX_LENGTH
                ),
                hashes.SHA256()
            )
            return True
        except Exception:
            return False

    def check_freshness(self, timestamp: float, max_age_seconds: int = 60) -> bool:
        """Ensures the attestation is not a replay attack."""
        return (time.time() - timestamp) < max_age_seconds
Functions
check_freshness(timestamp, max_age_seconds=60)

Ensures the attestation is not a replay attack.

Source code in src/safety/security.py 
33
34
35
def check_freshness(self, timestamp: float, max_age_seconds: int = 60) -> bool:
    """Ensures the attestation is not a replay attack."""
    return (time.time() - timestamp) < max_age_seconds
verify_runtime_attestation(attestation_report, signature)

Verifies that the runtime attestation report was signed by the trusted platform (e.g. simulated TPM or Cloud HSM). NOTE: In production, the public key should be fetched from GCP Secret Manager or a KMS-backed Identity Provider with automated rotation enabled.

Source code in src/safety/security.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def verify_runtime_attestation(self, attestation_report: bytes, signature: bytes) -> bool:
    """
    Verifies that the runtime attestation report was signed by the 
    trusted platform (e.g. simulated TPM or Cloud HSM).
    NOTE: In production, the public key should be fetched from GCP Secret Manager 
    or a KMS-backed Identity Provider with automated rotation enabled.
    """
    try:
        self.public_key.verify(
            signature,
            attestation_report,
            padding.PSS(
                mgf=padding.MGF1(hashes.SHA256()),
                salt_length=padding.PSS.MAX_LENGTH
            ),
            hashes.SHA256()
        )
        return True
    except Exception:
        return False