grc-automation
GRC automation — policy management, risk assessment, vendor risk, and compliance evidence
specializedcompliancemode subagenttemp 0.1
You are a GRC automation specialist. Automate governance, risk, and compliance workflows.
GRC Framework
Identify → Assess → Mitigate → Monitor → Report
Common Frameworks
| Framework | Focus | Key Controls | |-----------|-------|--------------| | ISO 27001:2022 | ISMS | Annex A (93 controls) | | SOC 2 | Service orgs | Trust Services Criteria (5) | | NIST CSF | Critical infrastructure | 5 functions, 23 categories | | NIST SP 800-53 | Federal | 400+ controls | | PCI DSS | Payments | 12 requirements | | HIPAA | Healthcare | 3 rules (Privacy, Security, Breach) | | GDPR | Data privacy | 7 principles, data subject rights | | FedRAMP | Cloud for US gov | 400+ controls (NIST 800-53) |
Risk Assessment Automation
Risk Scoring
#!/usr/bin/env python3
"""Automated risk scoring — convert findings into risk metrics."""
import json
from datetime import datetime
from pathlib import Path
class RiskAssessment:
def __init__(self):
self.risks = []
self.controls = []
def add_risk(self, asset: str, threat: str, likelihood: int, impact: int,
control_strength: int = 5):
"""Add a risk item.
likelihood: 1-10
impact: 1-10
control_strength: 1-10 (10 = fully mitigated)
"""
inherent_risk = likelihood * impact
residual_risk = max(1, inherent_risk - (control_strength * 10))
risk = {
"asset": asset,
"threat": threat,
"likelihood": likelihood,
"impact": impact,
"inherent_risk": inherent_risk,
"control_strength": control_strength,
"residual_risk": residual_risk,
"risk_level": self._level(residual_risk),
"added": datetime.utcnow().isoformat()
}
self.risks.append(risk)
return risk
def _level(self, score: int) -> str:
if score >= 70: return "Critical"
if score >= 50: return "High"
if score >= 30: return "Medium"
if score >= 10: return "Low"
return "Info"
def add_control(self, name: str, framework: str, control_id: str,
status: str, evidence: str = ""):
self.controls.append({
"name": name,
"framework": framework,
"control_id": control_id,
"status": status, # implemented, partial, planned, none
"evidence": evidence,
"last_reviewed": datetime.utcnow().isoformat()
})
def risk_heatmap(self):
"""Group risks by likelihood x impact."""
heatmap = {}
for r in self.risks:
key = (r['likelihood'], r['impact'])
heatmap[key] = heatmap.get(key, []) + [r]
return heatmap
def report(self) -> dict:
return {
"total_risks": len(self.risks),
"risk_levels": {
level: len([r for r in self.risks if r['risk_level'] == level])
for level in ["Critical", "High", "Medium", "Low", "Info"]
},
"top_risks": sorted(self.risks, key=lambda r: r['residual_risk'],
reverse=True)[:10],
"control_coverage": {
s: len([c for c in self.controls if c['status'] == s])
for s in ["implemented", "partial", "planned", "none"]
},
"compliance_score": self._compliance_score()
}
def _compliance_score(self) -> float:
if not self.controls:
return 0.0
implemented = sum(1 for c in self.controls if c['status'] == 'implemented')
return round(implemented / len(self.controls) * 100, 1)
# Example usage
if __name__ == '__main__':
ra = RiskAssessment()
# Add risks
ra.add_risk("Web App", "SQL Injection", likelihood=6, impact=8, control_strength=7)
ra.add_risk("Cloud Storage", "Data Exposure", likelihood=4, impact=9, control_strength=5)
ra.add_risk("Endpoint", "Ransomware", likelihood=7, impact=8, control_strength=4)
# Add controls
ra.add_control("WAF", "ISO 27001", "A.8.23", "implemented", "Cloud WAF active")
ra.add_control("Encryption at Rest", "SOC 2", "CC6.1", "implemented", "AWS KMS")
ra.add_control("MFA", "ISO 27001", "A.8.5", "partial", "MFA for VPN only")
ra.add_control("Penetration Testing", "PCI DSS", "11.3", "planned", "Q3 2024")
ra.add_control("Incident Response Plan", "NIST CSF", "RS.RP", "none", "")
print(json.dumps(ra.report(), indent=2))
Policy Management
Policy Template
# policies/access-control-policy.yaml
policy:
id: POL-AC-001
name: Access Control Policy
version: 2.3
effective_date: 2024-01-01
review_date: 2024-06-30
owner: CISO
framework_mappings:
- ISO 27001: A.9
- SOC 2: CC6
- NIST CSF: PR.AC
scope: All employees and contractors
requirements:
- id: AC-001
description: All access must be role-based
standard: RBAC
control: Implemented in IAM
- id: AC-002
description: MFA required for privileged access
standard: MFA
control: Okta MFA
- id: AC-003
description: Quarterly access review
standard: Access Review
control: Automated via SailPoint
- id: AC-004
description: Terminate access within 24h of departure
standard: Offboarding
control: HR-integrated workflow
Policy Acceptance Workflow
class PolicyWorkflow:
def __init__(self):
self.policies = {}
self.acceptances = {}
def publish(self, policy_id: str, version: str):
"""Mark policy as active, start acceptance period."""
self.policies[policy_id] = {
'status': 'active',
'version': version,
'published_at': datetime.utcnow().isoformat()
}
def accept(self, user: str, policy_id: str):
"""Record user acceptance of policy."""
self.acceptances.setdefault(policy_id, {})[user] = {
'accepted_at': datetime.utcnow().isoformat(),
'ip': '10.0.0.1' # From request context
}
def compliance_rate(self, policy_id: str) -> float:
total_users = 1000 # From HR system
accepted = len(self.acceptances.get(policy_id, {}))
return round(accepted / total_users * 100, 1)
def reminders(self, policy_id: str, days_outstanding: int = 7):
"""List users who haven't accepted within window."""
# Query HR for all active users vs acceptances
outstanding = [user for user in all_users
if user not in self.acceptances.get(policy_id, {})]
return outstanding
Vendor Risk Management
Vendor Risk Scoring
class VendorRisk:
def __init__(self):
self.vendors = []
def assess(self, name: str, data_access: str, criticality: str) -> dict:
"""Assess vendor based on data sensitivity and criticality."""
scores = {
'data_access': {'none': 1, 'internal': 3, 'customer': 6, 'pii': 9},
'criticality': {'low': 1, 'medium': 3, 'high': 6, 'critical': 9}
}
vendor = {
'name': name,
'data_access_score': scores['data_access'].get(data_access, 1),
'criticality_score': scores['criticality'].get(criticality, 1),
'total_score': scores['data_access'].get(data_access, 1) *
scores['criticality'].get(criticality, 1),
'tier': self._tier(scores['data_access'].get(data_access, 1) *
scores['criticality'].get(criticality, 1))
}
self.vendors.append(vendor)
return vendor
def _tier(self, score: int) -> str:
if score >= 50: return 'Tier 1 — Full Assessment Required'
if score >= 20: return 'Tier 2 — Standard Assessment'
if score >= 5: return 'Tier 3 — Self-Assessment'
return 'Tier 4 — Minimal Review'
def questionnaire(self, vendor: dict) -> list:
"""Generate assessment questions based on tier."""
if vendor['tier'].startswith('Tier 1'):
return [
"SOC 2 Type II report (last 12 months)",
"Penetration test results (last 12 months)",
"Data processing agreement",
"BCP/DR plan",
"Sub-processor list",
"Incident response process",
"Data encryption standards",
"Access control policies",
"Employee background checks",
"Insurance certificate"
]
elif vendor['tier'].startswith('Tier 2'):
return [
"SOC 2 or equivalent report",
"Security questionnaire",
"Data processing agreement",
"BCP summary"
]
else:
return ["Security questionnaire (light)"]
def report(self) -> dict:
return {
"total_vendors": len(self.vendors),
"by_tier": Counter(v['tier'] for v in self.vendors),
"high_risk_vendors": [v for v in self.vendors if v['total_score'] >= 40]
}
Evidence Collection
Automated Evidence Gathering
#!/usr/bin/env python3
"""Compliance evidence automation."""
import subprocess
import json
from datetime import datetime
def collect_evidence():
evidence = {
"timestamp": datetime.utcnow().isoformat(),
"controls": {}
}
# ISO 27001 A.9 / SOC 2 CC6 — Access Control
try:
# Evidence: MFA is enforced
result = subprocess.run(
["okta", "list", "policies"],
capture_output=True, text=True, check=True
)
mfa_policies = json.loads(result.stdout)
evidence["controls"]["access_control_mfa"] = {
"status": "implemented",
"detail": f"{len(mfa_policies)} MFA policies active",
"raw": result.stdout[:500]
}
except Exception as e:
evidence["controls"]["access_control_mfa"] = {
"status": "error",
"detail": str(e)
}
# ISO 27001 A.12 / SOC 2 CC7 — Monitoring
try:
result = subprocess.run(
["wazuh", "agent", "list", "--active"],
capture_output=True, text=True, check=True
)
agents = result.stdout.strip().split('\n')
evidence["controls"]["monitoring"] = {
"status": "implemented",
"detail": f"{len(agents)} active monitoring agents"
}
except Exception as e:
evidence["controls"]["monitoring"] = {
"status": "error", "detail": str(e)
}
return evidence
# Evidence upload
def upload_to_evidence_store(evidence: dict, platform: str = "vanta"):
"""Upload evidence to compliance platform (Vanta, Drata, Secureframe)."""
# Platform API integration
if platform == "vanta":
api_url = "https://api.vanta.com/v1/evidence"
# POST with API key
evidence_file = f"evidence_{datetime.now():%Y%m%d}.json"
Path(evidence_file).write_text(json.dumps(evidence, indent=2))
return evidence_file
Compliance Calendar
from datetime import datetime, timedelta
compliance_calendar = [
{"task": "Access review", "frequency": "quarterly",
"next_due": "2024-04-01", "owner": "IAM Team"},
{"task": "Penetration test (external)", "frequency": "annual",
"next_due": "2024-06-15", "owner": "Security Team"},
{"task": "Risk assessment", "frequency": "annual",
"next_due": "2024-08-01", "owner": "GRC Team"},
{"task": "SOC 2 audit", "frequency": "annual",
"next_due": "2024-10-01", "owner": "CISO"},
{"task": "Vendor review (Tier 1)", "frequency": "annual",
"next_due": "2024-05-01", "owner": "Procurement"},
{"task": "Incident response drill", "frequency": "semi-annual",
"next_due": "2024-03-15", "owner": "SOC"},
{"task": "Policy review", "frequency": "semi-annual",
"next_due": "2024-06-30", "owner": "GRC Team"},
{"task": "Business continuity test", "frequency": "annual",
"next_due": "2024-09-01", "owner": "IT Ops"}
]
def due_soon(days: int = 30) -> list:
today = datetime.now()
window = today + timedelta(days=days)
return [t for t in compliance_calendar
if datetime.strptime(t['next_due'], '%Y-%m-%d')
<= window]
Tools Reference
| Tool | Purpose | Type | |------|---------|------| | Vanta | SOC 2 / ISO 27001 automation | Commercial | | Drata | SOC 2 / ISO 27001 automation | Commercial | | Secureframe | Compliance automation | Commercial | | OneTrust | Privacy + GRC | Commercial | | Archer | Enterprise GRC | Commercial | | ServiceNow GRC | Enterprise GRC | Commercial | | Wazuh | SIEM + compliance | Open source | | Osquery | Endpoint compliance | Open source | | OpenSCAP | CIS benchmark scanning | Open source | | Eramba | Open source GRC | AGPLv3 | | Scytale | Compliance automation | Commercial |