feat(tyr): implement eval_tyr_identity and get_tyr_user_risk tools

project/roadmap.md

@@ -29,6 +29,7 @@ This document tracks the high-level goals and future development milestones for
 
 ## Phase 5: OISSU Loop & Custom MCP Security Tools
 - [x] Build OISSU forecast and anomaly detection tools.
+- [x] Build `eval_tyr_identity` and `get_tyr_user_risk` tools.
 - [ ] 5.2 Build `eval_tyr_identity` and `get_tyr_user_risk` tools.
 - [ ] 5.3 Build `scan_tyr_surface` and `attest_tyr_supply_chain` tools.
 - [ ] 5.4 Build `run_tyr_response` tool for automated threat containment.

tyr/memory_bank/completed.md

@@ -51,3 +51,7 @@
 - **Task 5.1: Implement OISSU Tools**
   - Status: **Complete**
   - Notes: Implemented initial logic for `get_tyr_forecast` and `scan_tyr_surface` tools.
+
+- **Task 5.2: Implement Identity & Risk Tools**
+  - Status: **Complete**
+  - Notes: Implemented `eval_tyr_identity` and `get_tyr_user_risk` tools.

tyr/tools/eval_tyr_identity.py

@@ -0,0 +1,31 @@
+#!/usr/bin/env python
+#
+# tyr/tools/eval_tyr_identity.py - MCP Tool for verifying SPIFFE identities
+#
+import subprocess
+
+def eval_tyr_identity(p: dict) -> dict:
+    """Verifies the validity and properties of a SPIFFE SVID by querying the SPIRE server."""
+    spiffe_id = p.get("spiffe_id")
+    if not spiffe_id:
+        raise ValueError("Missing required parameter: 'spiffe_id'")
+
+    if not spiffe_id.startswith("spiffe://vauco.no/"):
+        return {"status": "FAIL", "reason": "Invalid SPIFFE ID format for trust domain 'vauco.no'."}
+
+    try:
+        # This assumes the spire-server binary is in the path or at this location
+        cmd = ["./spire-1.15.3/bin/spire-server", "entry", "show", "-spiffeID", spiffe_id]
+        result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=10)
+        
+        if "Found 1 entry" in result.stdout:
+             # A more robust check would parse the TTL and other details
+             return {"status": "PASS", "spiffe_id": spiffe_id, "message": "Entry found and is active."}
+        else:
+             return {"status": "FAIL", "spiffe_id": spiffe_id, "reason": "SPIFFE ID not found or has no active entry."}
+
+    except subprocess.CalledProcessError as e:
+        # This handles cases where the entry is not found, as spire-server returns a non-zero exit code
+        return {"status": "FAIL", "spiffe_id": spiffe_id, "reason": f"SPIRE server error: Entry not found."}
+    except Exception as e:
+        return {"status": "ERROR", "message": str(e)}

tyr/tools/get_tyr_user_risk.py

@@ -0,0 +1,77 @@
+#!/usr/bin/env python
+#
+# tyr/tools/get_tyr_user_risk.py - MCP Tool for calculating user risk scores
+#
+import os
+from google.cloud import bigquery
+from datetime import datetime, timedelta
+
+RISK_WEIGHTS = {
+    "google.cloud.secretmanager.v1.SecretManagerService.AccessSecretVersion": 10,
+    "google.iam.admin.v1.SetIamPolicy": 20,
+    "google.storage.v1.Storage.DeleteBucket": 25,
+    "google.login.LoginService.LoginFailure": 5,
+}
+
+def get_tyr_user_risk(p: dict) -> dict:
+    """Calculates a risk score for a user based on recent audit log activity."""
+    project_id = os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5")
+    principal_email = p.get("principal_email")
+    caller_ip = p.get("caller_ip")
+    hours = p.get("hours", 24)
+
+    if not (principal_email or caller_ip):
+        raise ValueError("Missing required parameter: 'principal_email' or 'caller_ip'")
+
+    client = bigquery.Client()
+    table_id = f"{project_id}.tyr_audit_logs.cloudaudit_googleapis_com_activity"
+    
+    where_clauses = [f"timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {hours} HOUR)"]
+    if principal_email:
+        where_clauses.append(f"protopayload_auditlog.authenticationInfo.principalEmail = '{principal_email}'")
+    if caller_ip:
+        where_clauses.append(f"protopayload_auditlog.requestMetadata.callerIp = '{caller_ip}'")
+    
+    query = f"""
+        SELECT
+            timestamp,
+            protopayload_auditlog.methodName,
+            protopayload_auditlog.status.message as status_message
+        FROM `{table_id}`
+        WHERE {" AND ".join(where_clauses)}
+    """
+
+    try:
+        query_job = client.query(query)
+        results = query_job.result()
+        
+        risk_score = 0
+        findings = []
+        
+        for row in results:
+            weight = RISK_WEIGHTS.get(row.methodName, 0) # Default to 0 for non-risky methods
+            if row.status_message and "authentication failed" in row.status_message.lower():
+                weight = RISK_WEIGHTS.get("google.login.LoginService.LoginFailure", 5)
+            
+            if weight > 0:
+                risk_score += weight
+                findings.append({
+                    "timestamp": row.timestamp.isoformat(),
+                    "method": row.methodName,
+                    "status": row.status_message,
+                    "weight": weight
+                })
+
+        return {
+            "status": "success",
+            "principal_email": principal_email,
+            "caller_ip": caller_ip,
+            "time_window_hours": hours,
+            "calculated_risk_score": risk_score,
+            "findings_count": len(findings),
+            "findings": findings
+        }
+
+    except Exception as e:
+        print(f"An error occurred while querying audit logs: {e}")
+        raise