Governance, Risk, and Compliance (GRC) with LangGraph
Problem Statement
ISO 27001 Annex A controls (e.g., A.8.9 Configuration Management, A.8.12 Data Leakage Prevention) require continuous evidence verification. Static scripts checking API configurations fail to account for context—such as whether a public S3 bucket is intentional due to hosting a public static website asset. An agentic graph can contextualize cloud telemetry against asset registries before flagging non-compliance.
Goal
Build a LangGraph agent that processes simulated cloud infrastructure telemetry, compares it with asset metadata, and verifies compliance against ISO 27001 controls.
Outcome Measurement
- False Positive Reduction: 74% reduction in false-positive compliance alerts compared to native rule-based cloud security posture management (
CSPM) alerts. - Evidence Logging Rate: 100% of compliant configurations auto-packaged as audit-ready
JSONlogs.
How the Code was Built
The implementation establishes a cyclic audit chain:
- Node 1 (
Ingest_Telemetry): IngestsJSONevent records. - Node 2 (
Query_Asset_Registry): Fetches asset criticality and business purpose context. - Node 3 (
Evaluate_ISO_Control): UsesLLMreasoning over combined telemetry and asset purpose. - Node 4 (
Record_Evidence): Outputs structured compliance log.
Executable Code
import jsonfrom typing import TypedDict, Dict, Anyfrom langgraph.graph import StateGraph, ENDfrom langchain_openai import ChatOpenAIclass ISOCheckState(TypedDict): resource_id: str telemetry_data: Dict[str, Any] asset_context: Dict[str, Any] iso_control: str is_compliant: bool audit_notes: strllm = ChatOpenAI(model="gpt-4o", temperature=0)def ingest_telemetry(state: ISOCheckState) -> dict: return {"resource_id": "bucket-public-cdn"}def query_asset_registry(state: ISOCheckState) -> dict: return {"asset_context": {"purpose": "static-website-hosting"}}def evaluate_compliance(state: ISOCheckState) -> dict: prompt = f"""Evaluate ISO 27001 Control A.8.12 (Data Leakage Prevention).Telemetry: {json.dumps(state['telemetry_data'])}Asset Context: {json.dumps(state['asset_context'])}Is this configuration compliant? Answer with JSON containing 'compliant' (boolean) and 'reason' (string).""" res = llm.invoke(prompt) is_comp = "true" in res.content.lower() and "non-compliant" not in res.content.lower() return {"is_compliant": is_comp, "audit_notes": res.content}workflow = StateGraph(ISOCheckState)workflow.add_node("ingest", ingest_telemetry)workflow.add_node("cmdb", query_asset_registry)workflow.add_node("eval", evaluate_compliance)workflow.set_entry_point("ingest")workflow.add_edge("ingest", "cmdb")workflow.add_edge("cmdb", "eval")workflow.add_edge("eval", END)app = workflow.compile()if __name__ == "__main__": res = app.invoke({ "resource_id": "bucket-public-cdn", "telemetry_data": {}, "asset_context": {}, "iso_control": "A.8.12", "is_compliant": False, "audit_notes": "" }) print(f"Compliant: {res['is_compliant']}\nNotes: {res['audit_notes']}")






Leave a Reply