81 lines
3.3 KiB
Python
81 lines
3.3 KiB
Python
#!/usr/bin/env python
|
|
#
|
|
# tyr/tools/scan_tyr_surface.py - MCP Tool for security surface scanning
|
|
#
|
|
import os
|
|
import json
|
|
import subprocess
|
|
from datetime import datetime
|
|
from google.cloud import secretmanager
|
|
|
|
|
|
def scan_tyr_surface(p: dict) -> dict:
|
|
"""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
|