Course: 2A — Building AI Harnesses for Cybersecurity Module: S09 — Cloud Posture Harnesses Duration: 120–150 minutes (four labs, one per sub-section) Environment: Python 3.11+, Pydantic. Cloud CLI (AWS CLI / Azure CLI / gcloud) configured with read-only access to an isolated lab account only. No production credentials. A mock cloud inventory dataset (provided JSON) for offline development. An LLM API key for report generation (optional).
Safety boundary: All labs run against isolated cloud accounts or provided mock data. Never point the discovery harness at a production account. The lab account must be dedicated, disposable, and contain no real data.
from pydantic import BaseModel, Field
from typing import Literal
class CloudAsset(BaseModel):
asset_id: str
arn: str
provider: Literal["aws", "azure", "gcp"]
region: str
account_id: str
service: str
resource_type: str # normalized: "storage", "compute", "identity", "ai_model"
is_ai_asset: bool = False
ai_tags: dict | None = None # {model_name, endpoint_publicly_reachable, vector_store, ...}
is_public: bool = False
attached_policies: list[str] = Field(default_factory=list)
network_exposure: Literal["public", "vpc", "private"] = "private"
encryption_at_rest: bool = True
encryption_in_transit: bool = True
trust_relationships: list[dict] = Field(default_factory=list)
data_classification: Literal["public", "internal", "sensitive", "regulated"] = "internal"
last_assessed: str = ""
import json
async def discover_aws_assets(profile: str = "lab") -> list[CloudAsset]:
"""Enumerate AWS resources. Use mock data or live lab account (read-only)."""
# Option A: load provided mock dataset
with open("mock-aws-inventory.json") as f:
raw = json.load(f)
return [CloudAsset(**normalize_aws(r)) for r in raw["resources"]]
# Option B: live lab account (read-only, isolated)
# import boto3
# session = boto3.Session(profile_name=profile)
# ... enumerate via Config / Security Hub APIs ...
def normalize_aws(raw: dict) -> dict:
"""Normalize raw AWS resource to CloudAsset schema."""
return {
"asset_id": raw["ARN"],
"arn": raw["ARN"],
"provider": "aws",
"region": raw.get("Region", "us-east-1"),
"account_id": raw.get("AccountId", ""),
"service": raw.get("ResourceType", "").split("::")[0].lower() if "::" in raw.get("ResourceType","") else "unknown",
"resource_type": map_aws_type(raw.get("ResourceType", "")),
"is_public": raw.get("IsPublic", False),
"is_ai_asset": is_ai_resource(raw),
"ai_tags": extract_ai_tags(raw),
"network_exposure": "public" if raw.get("IsPublic") else "private",
"data_classification": raw.get("DataClassification", "internal"),
}
AI_RESOURCE_PATTERNS = {
"aws:sagemaker:endpoint": {"is_ai": True, "type": "ai_model"},
"aws:bedrock:agent": {"is_ai": True, "type": "ai_agent"},
"aws:lambda:function": {"is_ai": False, "type": "compute"}, # may host AI, check tags
}
def is_ai_resource(raw: dict) -> bool:
rtype = raw.get("ResourceType", "").lower()
for pattern in AI_RESOURCE_PATTERNS:
if pattern in rtype:
return AI_RESOURCE_PATTERNS[pattern]["is_ai"]
# Check tags for AI markers
tags = raw.get("Tags", {})
return any(k.lower() in ("ai", "ml", "model", "llm") for k in tags)
def extract_ai_tags(raw: dict) -> dict | None:
if not is_ai_resource(raw):
return None
return {
"endpoint_publicly_reachable": raw.get("IsPublic", False),
"model_name": raw.get("Tags", {}).get("ModelName"),
"vector_store": "vector" in raw.get("ResourceType", "").lower(),
}
async def reconcile_inventory(current: list[CloudAsset], last_known: dict[str, CloudAsset]) -> dict:
"""Diff current state against last-known inventory."""
current_map = {a.asset_id: a for a in current}
return {
"new": [a for aid, a in current_map.items() if aid not in last_known],
"changed": [a for aid, a in current_map.items()
if aid in last_known and a != last_known[aid]],
"deleted": [a for aid, a in last_known.items() if aid not in current_map],
}
from collections import defaultdict, deque
class AttackGraph:
def __init__(self, assets: list[CloudAsset]):
self.nodes = {a.asset_id: a for a in assets}
self.edges = defaultdict(list)
self._build_edges()
def _build_edges(self):
for asset in self.nodes.values():
for trust in asset.trust_relationships:
if trust["source_id"] in self.nodes:
self.edges[trust["source_id"]].append((asset.asset_id, trust["type"]))
# Public assets are entry points; data flow edges from exposure
if asset.is_public:
self.edges[asset.asset_id].append((asset.asset_id, "entry_point"))
def entry_points(self) -> list[str]:
return [a_id for a_id, a in self.nodes.items()
if a.network_exposure == "public" or a.is_public]
def sensitive_targets(self) -> list[str]:
return [a_id for a_id, a in self.nodes.items()
if a.data_classification in ("sensitive", "regulated")]
def find_paths(self, max_depth: int = 8) -> list[list[str]]:
"""BFS all paths from entry points to sensitive targets."""
paths = []
for entry in self.entry_points():
for target in self.sensitive_targets():
if entry == target:
continue
paths.extend(self._bfs(entry, target, max_depth))
return paths
def _bfs(self, start: str, goal: str, max_depth: int) -> list[list[str]]:
results = []
queue = deque([(start, [start])])
while queue:
node, path = queue.popleft()
if len(path) > max_depth:
continue
for neighbor, edge_type in self.edges.get(node, []):
if neighbor == goal:
results.append(path + [neighbor])
elif neighbor not in path: # avoid cycles
queue.append((neighbor, path + [neighbor]))
return results
def score_path(self, path: list[str]) -> float:
score = 0.0
for node_id in path:
node = self.nodes[node_id]
if node.data_classification in ("sensitive", "regulated"):
score += 40
if node.network_exposure == "public":
score += 10
if node.service == "iam" and len(node.attached_policies) > 3:
score += 15 # over-privileged heuristic
if not node.encryption_at_rest:
score += 5
score *= (1.0 + (1.0 / len(path))) # shorter = more exploitable
return min(score, 100.0)
def top_attack_paths(self, n: int = 5) -> list[dict]:
paths = self.find_paths()
scored = [{"path": p, "score": self.score_path(p)} for p in paths]
return sorted(scored, key=lambda x: x["score"], reverse=True)[:n]
from dataclasses import dataclass
@dataclass
class CloudEvent:
event_id: str
timestamp: str
provider: str
event_source: str # e.g. "iam.amazonaws.com"
event_name: str # e.g. "AttachRolePolicy"
resource_arn: str
actor_arn: str
source_ip: str
SECURITY_RELEVANT = {
("iam", "AttachRolePolicy"), ("iam", "CreatePolicy"), ("iam", "PutRolePolicy"),
("iam", "AssumeRole"), ("iam", "CreateRole"),
("s3", "PutBucketAcl"), ("s3", "PutBucketPolicy"),
("ec2", "AuthorizeSecurityGroupIngress"), ("ec2", "ModifyInstanceAttribute"),
("sagemaker", "CreateEndpoint"), ("bedrock", "CreateAgent"),
}
def is_security_relevant(event: CloudEvent) -> bool:
source = event.event_source.split(".")[0]
return (source, event.event_name) in SECURITY_RELEVANT
async def posture_monitor(events: list[CloudEvent], graph: AttackGraph):
"""Process a stream of CloudTrail events; re-assess security-relevant changes."""
findings = []
for event in events:
if not is_security_relevant(event):
continue
# Re-fetch the specific resource's current state (mock: load from fixture)
asset = await fetch_resource_state(event.provider, event.resource_arn)
# Assess against benchmark controls
benchmark_findings = assess_against_cis(asset)
# Update the attack graph
graph.nodes[asset.asset_id] = asset
graph._build_edges()
# Score and route
for f in benchmark_findings:
f["exploitable"] = asset.asset_id in {n for path in graph.find_paths() for n in path}
f["severity"] = route_severity(f, asset)
findings.append(f)
return findings
def route_severity(finding: dict, asset: CloudAsset) -> str:
if asset.data_classification == "regulated" and asset.is_public:
return "critical"
if asset.is_public:
return "high"
if asset.data_classification in ("sensitive", "regulated"):
return "medium"
return "low"
mock-cloudtrail-events.json (50 events, ~15 security-relevant).def generate_ciso_report(graph: AttackGraph, findings: list[dict], history: list[dict]) -> dict:
top_paths = graph.top_attack_paths(n=5)
return {
"risk_score": compute_overall_risk(findings),
"risk_trend": trend_direction(history, window_days=30),
"top_attack_paths": [
{
"path": [graph.nodes[n].resource_type for n in p["path"]],
"score": round(p["score"], 1),
"target_classification": graph.nodes[p["path"][-1]].data_classification,
}
for p in top_paths
],
"critical_findings_open": sum(1 for f in findings if f["severity"] == "critical"),
"compliance_posture": compliance_summary(findings),
"headline": generate_headline(findings, top_paths),
}
def compute_overall_risk(findings: list[dict]) -> float:
weights = {"critical": 25, "high": 10, "medium": 3, "low": 1}
return min(100.0, sum(weights.get(f["severity"], 1) for f in findings))
from typing import Literal
class RemediationPolicy(BaseModel):
environment: Literal["production", "staging", "development", "sandbox"]
remediation_type: str
auto_apply: bool
requires_approval_from: list[str]
rollback_plan: str
def evaluate_remediation(
finding: dict,
environment: str,
approver: str | None,
) -> Literal["propose_only", "apply", "blocked"]:
# Production ALWAYS requires approval — the load-bearing rule
if environment == "production" and approver is None:
return "propose_only"
if environment == "production":
return "apply" # approver present
# Non-production: auto-apply per policy
if environment in ("staging", "development", "sandbox"):
return "apply"
return "blocked"
def create_remediation_ticket(finding: dict, action: str) -> dict:
return {
"title": f"[{finding['severity'].upper()}] {finding['control_id']} on {finding['resource_arn']}",
"action": action,
"remediation_command": generate_remediation_command(finding),
"approval_required": action == "propose_only",
"rollback_plan": generate_rollback_plan(finding),
}
production and sandbox environments.propose_only; sandbox findings are apply.# Lab Specification — Module S09: Cloud Posture Harnesses
**Course**: 2A — Building AI Harnesses for Cybersecurity
**Module**: S09 — Cloud Posture Harnesses
**Duration**: 120–150 minutes (four labs, one per sub-section)
**Environment**: Python 3.11+, Pydantic. Cloud CLI (AWS CLI / Azure CLI / gcloud) configured with read-only access to an **isolated lab account only**. No production credentials. A mock cloud inventory dataset (provided JSON) for offline development. An LLM API key for report generation (optional).
> **Safety boundary**: All labs run against isolated cloud accounts or provided mock data. Never point the discovery harness at a production account. The lab account must be dedicated, disposable, and contain no real data.
---
## Learning objectives
1. Build a cloud asset discovery harness that enumerates resources across one cloud provider and produces a structured inventory with AI workload tagging.
2. Build an attack path analyzer that takes a cloud asset graph and finds all paths from externally reachable assets to sensitive data stores.
3. Build an event-driven posture monitor that processes a stream of CloudTrail events and triggers targeted re-assessments on security-relevant changes.
4. Generate a CISO-ready risk summary from finding data and open remediation tickets for high-severity findings behind an approval gate.
---
## Phase 1 — Cloud Asset Discovery Harness (35 min)
### 1.1 Define the CloudAsset schema
```python
from pydantic import BaseModel, Field
from typing import Literal
class CloudAsset(BaseModel):
asset_id: str
arn: str
provider: Literal["aws", "azure", "gcp"]
region: str
account_id: str
service: str
resource_type: str # normalized: "storage", "compute", "identity", "ai_model"
is_ai_asset: bool = False
ai_tags: dict | None = None # {model_name, endpoint_publicly_reachable, vector_store, ...}
is_public: bool = False
attached_policies: list[str] = Field(default_factory=list)
network_exposure: Literal["public", "vpc", "private"] = "private"
encryption_at_rest: bool = True
encryption_in_transit: bool = True
trust_relationships: list[dict] = Field(default_factory=list)
data_classification: Literal["public", "internal", "sensitive", "regulated"] = "internal"
last_assessed: str = ""
```
### 1.2 Implement provider discovery (mock or live lab account)
```python
import json
async def discover_aws_assets(profile: str = "lab") -> list[CloudAsset]:
"""Enumerate AWS resources. Use mock data or live lab account (read-only)."""
# Option A: load provided mock dataset
with open("mock-aws-inventory.json") as f:
raw = json.load(f)
return [CloudAsset(**normalize_aws(r)) for r in raw["resources"]]
# Option B: live lab account (read-only, isolated)
# import boto3
# session = boto3.Session(profile_name=profile)
# ... enumerate via Config / Security Hub APIs ...
def normalize_aws(raw: dict) -> dict:
"""Normalize raw AWS resource to CloudAsset schema."""
return {
"asset_id": raw["ARN"],
"arn": raw["ARN"],
"provider": "aws",
"region": raw.get("Region", "us-east-1"),
"account_id": raw.get("AccountId", ""),
"service": raw.get("ResourceType", "").split("::")[0].lower() if "::" in raw.get("ResourceType","") else "unknown",
"resource_type": map_aws_type(raw.get("ResourceType", "")),
"is_public": raw.get("IsPublic", False),
"is_ai_asset": is_ai_resource(raw),
"ai_tags": extract_ai_tags(raw),
"network_exposure": "public" if raw.get("IsPublic") else "private",
"data_classification": raw.get("DataClassification", "internal"),
}
```
### 1.3 AI workload tagging
```python
AI_RESOURCE_PATTERNS = {
"aws:sagemaker:endpoint": {"is_ai": True, "type": "ai_model"},
"aws:bedrock:agent": {"is_ai": True, "type": "ai_agent"},
"aws:lambda:function": {"is_ai": False, "type": "compute"}, # may host AI, check tags
}
def is_ai_resource(raw: dict) -> bool:
rtype = raw.get("ResourceType", "").lower()
for pattern in AI_RESOURCE_PATTERNS:
if pattern in rtype:
return AI_RESOURCE_PATTERNS[pattern]["is_ai"]
# Check tags for AI markers
tags = raw.get("Tags", {})
return any(k.lower() in ("ai", "ml", "model", "llm") for k in tags)
def extract_ai_tags(raw: dict) -> dict | None:
if not is_ai_resource(raw):
return None
return {
"endpoint_publicly_reachable": raw.get("IsPublic", False),
"model_name": raw.get("Tags", {}).get("ModelName"),
"vector_store": "vector" in raw.get("ResourceType", "").lower(),
}
```
### 1.4 Reconciliation loop
```python
async def reconcile_inventory(current: list[CloudAsset], last_known: dict[str, CloudAsset]) -> dict:
"""Diff current state against last-known inventory."""
current_map = {a.asset_id: a for a in current}
return {
"new": [a for aid, a in current_map.items() if aid not in last_known],
"changed": [a for aid, a in current_map.items()
if aid in last_known and a != last_known[aid]],
"deleted": [a for aid, a in last_known.items() if aid not in current_map],
}
```
### Deliverable
- [ ] CloudAsset schema implemented with all fields
- [ ] Discovery from mock dataset (or live lab account, read-only) normalized to schema
- [ ] AI workload tagging correctly flags model endpoints, agents, vector DBs
- [ ] Reconciliation loop detects new, changed, and deleted assets
- [ ] Verify: no production credentials used; isolated lab account only
---
## Phase 2 — Attack Path Analyzer (35 min)
### 2.1 Build the attack graph
```python
from collections import defaultdict, deque
class AttackGraph:
def __init__(self, assets: list[CloudAsset]):
self.nodes = {a.asset_id: a for a in assets}
self.edges = defaultdict(list)
self._build_edges()
def _build_edges(self):
for asset in self.nodes.values():
for trust in asset.trust_relationships:
if trust["source_id"] in self.nodes:
self.edges[trust["source_id"]].append((asset.asset_id, trust["type"]))
# Public assets are entry points; data flow edges from exposure
if asset.is_public:
self.edges[asset.asset_id].append((asset.asset_id, "entry_point"))
def entry_points(self) -> list[str]:
return [a_id for a_id, a in self.nodes.items()
if a.network_exposure == "public" or a.is_public]
def sensitive_targets(self) -> list[str]:
return [a_id for a_id, a in self.nodes.items()
if a.data_classification in ("sensitive", "regulated")]
```
### 2.2 Path-finding
```python
def find_paths(self, max_depth: int = 8) -> list[list[str]]:
"""BFS all paths from entry points to sensitive targets."""
paths = []
for entry in self.entry_points():
for target in self.sensitive_targets():
if entry == target:
continue
paths.extend(self._bfs(entry, target, max_depth))
return paths
def _bfs(self, start: str, goal: str, max_depth: int) -> list[list[str]]:
results = []
queue = deque([(start, [start])])
while queue:
node, path = queue.popleft()
if len(path) > max_depth:
continue
for neighbor, edge_type in self.edges.get(node, []):
if neighbor == goal:
results.append(path + [neighbor])
elif neighbor not in path: # avoid cycles
queue.append((neighbor, path + [neighbor]))
return results
```
### 2.3 Score paths
```python
def score_path(self, path: list[str]) -> float:
score = 0.0
for node_id in path:
node = self.nodes[node_id]
if node.data_classification in ("sensitive", "regulated"):
score += 40
if node.network_exposure == "public":
score += 10
if node.service == "iam" and len(node.attached_policies) > 3:
score += 15 # over-privileged heuristic
if not node.encryption_at_rest:
score += 5
score *= (1.0 + (1.0 / len(path))) # shorter = more exploitable
return min(score, 100.0)
def top_attack_paths(self, n: int = 5) -> list[dict]:
paths = self.find_paths()
scored = [{"path": p, "score": self.score_path(p)} for p in paths]
return sorted(scored, key=lambda x: x["score"], reverse=True)[:n]
```
### Deliverable
- [ ] Attack graph built from CloudAsset inventory (nodes + edges)
- [ ] BFS path-finding from entry points to sensitive targets
- [ ] Path scoring (target value, exposure, privilege, length)
- [ ] Top 5 attack paths identified from the provided mock inventory
- [ ] Verify: fixing one node (e.g., removing a trust edge) breaks paths through it
---
## Phase 3 — Event-Driven Posture Monitor (35 min)
### 3.1 Define the CloudEvent and security-relevant filter
```python
from dataclasses import dataclass
@dataclass
class CloudEvent:
event_id: str
timestamp: str
provider: str
event_source: str # e.g. "iam.amazonaws.com"
event_name: str # e.g. "AttachRolePolicy"
resource_arn: str
actor_arn: str
source_ip: str
SECURITY_RELEVANT = {
("iam", "AttachRolePolicy"), ("iam", "CreatePolicy"), ("iam", "PutRolePolicy"),
("iam", "AssumeRole"), ("iam", "CreateRole"),
("s3", "PutBucketAcl"), ("s3", "PutBucketPolicy"),
("ec2", "AuthorizeSecurityGroupIngress"), ("ec2", "ModifyInstanceAttribute"),
("sagemaker", "CreateEndpoint"), ("bedrock", "CreateAgent"),
}
def is_security_relevant(event: CloudEvent) -> bool:
source = event.event_source.split(".")[0]
return (source, event.event_name) in SECURITY_RELEVANT
```
### 3.2 Process the event stream
```python
async def posture_monitor(events: list[CloudEvent], graph: AttackGraph):
"""Process a stream of CloudTrail events; re-assess security-relevant changes."""
findings = []
for event in events:
if not is_security_relevant(event):
continue
# Re-fetch the specific resource's current state (mock: load from fixture)
asset = await fetch_resource_state(event.provider, event.resource_arn)
# Assess against benchmark controls
benchmark_findings = assess_against_cis(asset)
# Update the attack graph
graph.nodes[asset.asset_id] = asset
graph._build_edges()
# Score and route
for f in benchmark_findings:
f["exploitable"] = asset.asset_id in {n for path in graph.find_paths() for n in path}
f["severity"] = route_severity(f, asset)
findings.append(f)
return findings
def route_severity(finding: dict, asset: CloudAsset) -> str:
if asset.data_classification == "regulated" and asset.is_public:
return "critical"
if asset.is_public:
return "high"
if asset.data_classification in ("sensitive", "regulated"):
return "medium"
return "low"
```
### 3.3 Run against provided event stream
1. Load `mock-cloudtrail-events.json` (50 events, ~15 security-relevant).
2. Load the initial asset inventory from Phase 1.
3. Process each event through the monitor.
4. Verify: only security-relevant events trigger re-assessment; others are logged only.
### Deliverable
- [ ] Security-relevant event filter implemented (IAM, S3, EC2, AI workload events)
- [ ] Event stream processed; only security-relevant events trigger re-assessment
- [ ] Findings routed by severity (exploitability × impact × reachability)
- [ ] Attack graph updated as events change resource state
- [ ] Verify: non-security-relevant events do NOT trigger re-assessment
---
## Phase 4 — CISO Report + Approval-Gated Remediation (35 min)
### 4.1 Generate the CISO risk summary
```python
def generate_ciso_report(graph: AttackGraph, findings: list[dict], history: list[dict]) -> dict:
top_paths = graph.top_attack_paths(n=5)
return {
"risk_score": compute_overall_risk(findings),
"risk_trend": trend_direction(history, window_days=30),
"top_attack_paths": [
{
"path": [graph.nodes[n].resource_type for n in p["path"]],
"score": round(p["score"], 1),
"target_classification": graph.nodes[p["path"][-1]].data_classification,
}
for p in top_paths
],
"critical_findings_open": sum(1 for f in findings if f["severity"] == "critical"),
"compliance_posture": compliance_summary(findings),
"headline": generate_headline(findings, top_paths),
}
def compute_overall_risk(findings: list[dict]) -> float:
weights = {"critical": 25, "high": 10, "medium": 3, "low": 1}
return min(100.0, sum(weights.get(f["severity"], 1) for f in findings))
```
### 4.2 Remediation with approval gate
```python
from typing import Literal
class RemediationPolicy(BaseModel):
environment: Literal["production", "staging", "development", "sandbox"]
remediation_type: str
auto_apply: bool
requires_approval_from: list[str]
rollback_plan: str
def evaluate_remediation(
finding: dict,
environment: str,
approver: str | None,
) -> Literal["propose_only", "apply", "blocked"]:
# Production ALWAYS requires approval — the load-bearing rule
if environment == "production" and approver is None:
return "propose_only"
if environment == "production":
return "apply" # approver present
# Non-production: auto-apply per policy
if environment in ("staging", "development", "sandbox"):
return "apply"
return "blocked"
def create_remediation_ticket(finding: dict, action: str) -> dict:
return {
"title": f"[{finding['severity'].upper()}] {finding['control_id']} on {finding['resource_arn']}",
"action": action,
"remediation_command": generate_remediation_command(finding),
"approval_required": action == "propose_only",
"rollback_plan": generate_rollback_plan(finding),
}
```
### 4.3 Run the full workflow
1. Take the findings from Phase 3.
2. Generate the CISO report (risk score, top 5 paths, headline).
3. For each high/critical finding, evaluate remediation under both `production` and `sandbox` environments.
4. Verify: production findings are `propose_only`; sandbox findings are `apply`.
5. Generate remediation tickets with full context.
### Deliverable
- [ ] CISO report generated (risk score, trend, top 5 attack paths, business-impact framing)
- [ ] Approval gate correctly routes production → propose_only, sandbox → apply
- [ ] Remediation tickets created with ARN, control, remediation command, rollback plan
- [ ] Verify: production findings NEVER auto-apply without approval
---
## Stretch goals
1. **Multi-provider normalization**: extend the discovery harness to ingest both AWS and Azure mock inventories and normalize them into one CloudAsset list. Verify cross-provider attack paths (e.g., an Azure identity with cross-cloud trust to an AWS role).
2. **Drift simulation**: simulate a sequence of CloudTrail events that introduce a new public S3 bucket, then attach a permissive policy, then create a trust relationship to a Lambda. Show how the attack graph evolves and a new path appears at each step.
3. **Auditor report view**: add a third report projection that maps each finding to a CIS control ID with pass/fail status, evidence timestamp, and assessor identity. Verify it derives from the same finding store as the CISO and engineer views.