chris/OSVauco
1
0
Fork 0
Code Issues 1 Pull Requests Actions 2 Packages Projects Releases Wiki Activity

feat(tyr): implement phase 5 anomaly detection and surface scanning tools

Browse Source
This commit is contained in:
Chris Christiansen 2026-09-02 20:00:47 +00:00
parent 55bb01f53c
commit 816bfc151b
5 changed files with 149 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
project/roadmap.md
View File

@ -22,12 +22,13 @@ This document tracks the high-level goals and future development milestones for
- [x] Secure Ollama model inference behind mTLS via SPIRE SVIDs.
- [x] Audit and sandbox execution environments using eBPF/Falco.
## Phase 3: Memory Bank & Project Management
- [x] Implement `read_memory_bank` and `write_memory_bank` MCP tools.
- [x] Implement `build_and_deploy_service` MCP tool.
- [ ] Implement `get_project_status` and `append_project_task` MCP tools.
## Phase 4: Data, Secret & CMEK Governance
- [x] Transition secrets to GCP Secret Manager
- [x] Enforce Customer-Managed Encryption Keys (CMEK) for Artifact Registry, Storage Buckets, and Cloud Run.
- [x] Configure BigQuery real-time audit log streaming and setup `query_tyr_audit` MCP tool.
## Phase 5: OISSU Loop & Custom MCP Security Tools
- [x] Build OISSU forecast and anomaly detection 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.

2
requirements.txt
View File

@ -53,3 +53,5 @@ authlib>=1.3.1
itsdangerous>=2.1.2
boto3>=1.34.120
google-api-python-client>=2.130.0
google-cloud-bigquery
google-cloud-secret-manager

4
tyr/memory_bank/completed.md
View File

@ -47,3 +47,7 @@
- **Task 4.3: Configure Audit Logging**
- Status: **Complete**
- Notes: Created BigQuery dataset and log sink for `cloudaudit.googleapis.com` logs.
- **Task 5.1: Implement OISSU Tools**
- Status: **Complete**
- Notes: Implemented initial logic for `get_tyr_forecast` and `scan_tyr_surface` tools.

76
tyr/tools/get_tyr_forecast.py
View File

@ -2,10 +2,78 @@
#
# tyr/tools/get_tyr_forecast.py - MCP Tool for anomaly detection
#
import os
from google.cloud import bigquery
from datetime import datetime, timedelta
import statistics
def get_tyr_forecast(p: dict) -> dict:
"""Analyzes audit log volume to find statistical anomalies."""
project_id = os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5")
hours_to_check = p.get("hours_to_check", 24)
std_dev_threshold = p.get("std_dev_threshold", 3.0)
client = bigquery.Client()
dataset_id = "tyr_audit_logs"
# This assumes a table partitioned by day, which is standard for log sinks.
# It queries the last 7 days to build a baseline.
query = f"""
SELECT
TIMESTAMP_TRUNC(timestamp, HOUR) as hour,
COUNT(*) as event_count
FROM `{project_id}.{dataset_id}.cloudaudit_googleapis_com_activity_*`
WHERE _TABLE_SUFFIX BETWEEN
FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE(), INTERVAL 8 DAY)) AND
FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE(), INTERVAL 1 DAY))
GROUP BY 1
ORDER BY 1
"""
Analyzes historical audit logs for anomalies using Vertex AI.
(Not yet implemented)
"""
return {"status": "not_implemented", "message": "Vertex AI IsolationForest/LSTM integration is pending."}
try:
query_job = client.query(query)
results = query_job.result()
hourly_counts = [row.event_count for row in results]
if len(hourly_counts) < 2:
return {"status": "error", "message": "Not enough historical data to calculate a baseline."}
# Calculate baseline mean and standard deviation
mean = statistics.mean(hourly_counts)
stdev = statistics.stdev(hourly_counts)
# Now, check the most recent N hours
query_recent = f"""
SELECT
TIMESTAMP_TRUNC(timestamp, HOUR) as hour,
COUNT(*) as event_count
FROM `{project_id}.{dataset_id}.cloudaudit_googleapis_com_activity_*`
WHERE timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL {hours_to_check} HOUR)
GROUP BY 1
ORDER BY 1 DESC
"""
query_job_recent = client.query(query_recent)
recent_results = query_job_recent.result()
anomalies = []
for row in recent_results:
if abs(row.event_count - mean) > (stdev * std_dev_threshold):
anomaly = {
"timestamp": row.hour.isoformat(),
"event_count": row.event_count,
"baseline_mean": round(mean, 2),
"baseline_stdev": round(stdev, 2),
"deviation": round((row.event_count - mean) / stdev, 2)
}
anomalies.append(anomaly)
if anomalies:
return {"status": "anomalies_found", "anomalies": anomalies}
else:
return {"status": "success", "message": f"No anomalies found in the last {hours_to_check} hours."}
except Exception as e:
print(f"An error occurred: {e}")
raise

70
tyr/tools/scan_tyr_surface.py
View File

@ -2,10 +2,70 @@
#
# tyr/tools/scan_tyr_surface.py - MCP Tool for security surface scanning
#
import os
import json
import subprocess
from google.cloud import secretmanager
def scan_tyr_surface(p: dict) -> dict:
"""
Scans for security vulnerabilities and misconfigurations.
(Not yet implemented)
"""
return {"status": "not_implemented", "message": "Surface scanning logic is pending."}
"""Scans GCP resources for compliance against key TYR security rules."""
project_id = os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5")
report = {
"scan_timestamp": datetime.utcnow().isoformat() + "Z",
"rules_checked": [],
"findings": []
}
# Rule Ω-SEC: Check Secret Manager rotation (< 30 days)
try:
report["rules_checked"].append("Ω-SEC")
client = secretmanager.SecretManagerServiceClient()
for secret in client.list_secrets(request={"parent": f"projects/{project_id}"}):
secret_details = client.get_secret(request={"name": secret.name})
rotation = secret_details.rotation
if not (rotation and rotation.rotation_period and rotation.rotation_period.seconds <= 2592000):
report["findings"].append({
"rule": "Ω-SEC",
"resource": secret_details.name,
"message": "Secret does not have a rotation period of 30 days or less."
})
except Exception as e:
report["findings"].append({"rule": "Ω-SEC", "status": "ERROR", "message": str(e)})
# Rule Ω-ID: Check SPIRE SVID TTLs (< 5 minutes)
try:
report["rules_checked"].append("Ω-ID")
# This is a simplified check. A full implementation would parse all entries.
cmd = ["./spire-1.15.3/bin/spire-server", "entry", "show"]
result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=10)
if "X509-SVID TTL : default" in result.stdout or "3600" in result.stdout:
report["findings"].append({
"rule": "Ω-ID",
"resource": "spire-server:default-ttl",
"message": "Default SVID TTL is in use (1 hour). It should be <= 5 minutes."
})
except Exception as e:
report["findings"].append({"rule": "Ω-ID", "status": "ERROR", "message": str(e)})
# Rule Ω-AUDIT: Check Log Sink
try:
report["rules_checked"].append("Ω-AUDIT")
cmd = ["gcloud", "logging", "sinks", "describe", "tyr-audit-sink", "--format=json"]
result = subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=10)
sink_info = json.loads(result.stdout)
if not sink_info.get("destination", "").endswith("datasets/tyr_audit_logs"):
report["findings"].append({
"rule": "Ω-AUDIT",
"resource": "tyr-audit-sink",
"message": "Log sink destination is not tyr_audit_logs."
})
if 'cloudaudit.googleapis.com' not in sink_info.get("filter", ""):
report["findings"].append({
"rule": "Ω-AUDIT",
"resource": "tyr-audit-sink",
"message": "Log sink is not configured to capture Cloud Audit Logs."
})
except Exception as e:
report["findings"].append({"rule": "Ω-AUDIT", "status": "ERROR", "message": str(e)})
return report