Course: 2A — Building AI Harnesses for Cybersecurity Module: S10 — Cloud Red Team Harnesses Duration: 90–120 minutes (three labs, one per sub-section) Environment: Python 3.11+, Pydantic. AWS CLI configured with read-only access to an isolated lab account only. Pacu and ScoutSuite installed (optional — mock data provided for offline). No production credentials. A mock IAM permission dataset (provided JSON). An LLM API key for report generation (optional).
Safety boundary: All labs run against an isolated, disposable lab account or provided mock data. Never point the offensive harness at a production account. The lab account must contain no real data. Scope enforcement must be active at all times.
from pydantic import BaseModel
from typing import Literal
class CloudScope(BaseModel):
engagement_id: str
cloud_provider: Literal["aws", "azure", "gcp"]
authorized_accounts: list[str] # account IDs in scope
allowed_regions: list[str]
denied_regions: list[str] # e.g. ["us-gov-*"]
blocked_services: list[str] # destructive: DeleteRole, DeleteBucket, TerminateInstances
max_api_calls_per_second: int = 10
LAB_SCOPE = CloudScope(
engagement_id="cloud-redteam-lab-001",
cloud_provider="aws",
authorized_accounts=["123456789012"], # lab account only
allowed_regions=["us-east-1", "us-west-2"],
denied_regions=["us-gov-*"],
blocked_services=[
"iam:DeleteRole", "iam:DeleteUser", "iam:DetachRolePolicy",
"s3:DeleteBucket", "s3:DeleteObject",
"ec2:TerminateInstances", "ec2:StopInstances",
],
)
import fnmatch
def check_scope(api_call: str, account_id: str, region: str, scope: CloudScope) -> tuple[bool, str]:
"""Check an API call against the engagement scope. Returns (allowed, reason)."""
# Account boundary
if account_id not in scope.authorized_accounts:
return False, f"BLOCKED: account {account_id} not in scope"
# Region check
if region not in scope.allowed_regions:
return False, f"BLOCKED: region {region} not allowed"
for denied in scope.denied_regions:
if fnmatch.fnmatch(region, denied):
return False, f"BLOCKED: region {region} matches denied pattern {denied}"
# Blocked services (destructive)
service_action = f"{api_call.split(':')[0]}:{api_call.split(':')[-1].split('.')[0]}"
for blocked in scope.blocked_services:
if blocked in api_call:
return False, f"BLOCKED: {api_call} is a blocked destructive action"
return True, "in scope"
async def enumerate_iam(scope: CloudScope, account_id: str, region: str) -> dict:
"""Enumerate IAM identities and permissions. Scope-checked."""
allowed, reason = check_scope("iam:ListRoles", account_id, region, scope)
if not allowed:
return {"blocked": True, "reason": reason}
# Execute (mock: load provided dataset, or live: boto3 call to lab account)
with open("mock-iam-permissions.json") as f:
import json
data = json.load(f)
# Evidence log every call
log_evidence(
api_call="iam:ListRoles", account_id=account_id, region=region,
scope_ref=f"{scope.engagement_id}:iam:ListRoles",
response_summary=f"{len(data['roles'])} roles enumerated",
)
return {"blocked": False, "roles": data["roles"], "policies": data["policies"]}
iam:DeleteRole; confirm blocked.ESCALATION_TECHNIQUES = [
{
"name": "PassRole + Lambda",
"requires": {"iam:PassRole", "lambda:CreateFunction", "lambda:InvokeFunction"},
"grants": "execution_role_permissions",
"description": "Pass a privileged role to a new Lambda, invoke it, act through the role",
},
{
"name": "CreatePolicy + AttachRolePolicy",
"requires": {"iam:CreatePolicy", "iam:AttachRolePolicy"},
"grants": "*:*",
"description": "Create a policy with full admin, attach to own role",
},
{
"name": "UpdateAssumeRolePolicy",
"requires": {"iam:UpdateAssumeRolePolicy"},
"grants": "target_role_permissions",
"description": "Modify a role's trust policy to permit assumption by current identity",
},
{
"name": "AssumeRole chain",
"requires": {"sts:AssumeRole"},
"grants": "assumed_role_permissions",
"description": "Assume a role that trusts the current identity",
},
]
from collections import deque
class PrivilegeGraph:
def __init__(self, identity_arn: str, permissions: set[str]):
self.start = identity_arn
self.permissions = permissions
def find_paths(self, target: set[str], max_depth: int = 5) -> list[dict]:
"""BFS from current permissions to target via escalation techniques."""
results = []
queue = deque([(list(self.permissions), [])])
visited = {frozenset(self.permissions)}
while queue:
current_perms, path = queue.popleft()
current_set = set(current_perms)
# Check if target reached
if target.issubset(current_set) or "*:*" in current_set:
results.append({"path": path, "final_permissions": current_perms})
if len(results) >= 10: # limit results
break
continue
if len(path) >= max_depth:
continue
for tech in ESCALATION_TECHNIQUES:
if tech["requires"].issubset(current_set):
# Simulate gaining the technique's grant
new_perms = current_perms + [tech["grants"]]
key = frozenset(new_perms)
if key not in visited:
visited.add(key)
queue.append((new_perms, path + [tech["name"]]))
return results
def available_techniques(self) -> list[str]:
"""Which escalation techniques are immediately available from current perms?"""
return [t["name"] for t in ESCALATION_TECHNIQUES
if t["requires"].issubset(self.permissions)]
# Load the starting identity's permissions from the mock dataset
with open("mock-iam-permissions.json") as f:
data = json.load(f)
starting_identity = data["identities"][0] # e.g. "arn:aws:sts::assumed-role/LowPrivRole"
starting_perms = set(starting_identity["permissions"])
graph = PrivilegeGraph(starting_identity["arn"], starting_perms)
# What techniques are available immediately?
print("Available techniques:", graph.available_techniques())
# Find all paths to admin (*:*)
target = {"*:*"}
paths = graph.find_paths(target)
for i, p in enumerate(paths):
print(f"Path {i+1}: {' -> '.join(p['path'])}")
import hashlib, json
from datetime import datetime, timezone
class CloudEvidenceRecord(BaseModel):
id: str
trace_id: str
session_id: str
timestamp: str
cloud_provider: str
account_id: str
region: str
api_call: str
resource_arn: str
actor_arn: str
request: str
response: str
http_status: int
scope_ref: str
finding_id: str | None = None
previous_hash: str = ""
record_hash: str = ""
retention_class: str = "restricted"
def create_evidence(data: dict, previous_hash: str) -> CloudEvidenceRecord:
record = {**data, "previous_hash": previous_hash, "timestamp": datetime.now(timezone.utc).isoformat()}
record_hash = hashlib.sha256(json.dumps(record, sort_keys=True).encode()).hexdigest()
record["record_hash"] = record_hash
return CloudEvidenceRecord(**record)
COMPLIANCE_MAP = {
"iam_over_privilege": {
"soc2": "CC6.3 — The entity authorizes, modifies, or removes access to data",
"iso27001": "A.9.2.5 — Review of user access rights",
"pci_dss": "7.2.1 — Establish an access control system for systems components",
},
"privilege_escalation_path": {
"soc2": "CC6.3 — Authorization of access (escalation path violates least privilege)",
"iso27001": "A.9.4.4 — Use of privileged utility programs",
"pci_dss": "7.1.1 — Define access needs for each role",
},
"public_exposure": {
"soc2": "CC6.1 — The entity implements logical access controls",
"iso27001": "A.13.1.1 — Network controls",
"pci_dss": "1.2.1 — Restrict inbound/outbound traffic",
},
}
def generate_client_report(findings: list[dict], evidence_chain: list[CloudEvidenceRecord]) -> dict:
"""Generate the 4-section client deliverable from findings + evidence."""
report = {
"executive_summary": {
"engagement_id": "cloud-redteam-lab-001",
"overall_risk": "HIGH — critical escalation paths identified",
"findings_count": len(findings),
"critical_findings": sum(1 for f in findings if f["severity"] == "critical"),
"business_impact": "Demonstrated paths from low-privilege identity to full admin access",
},
"finding_table": [
{
"id": f["id"],
"title": f["title"],
"severity": f["severity"],
"affected_resource": f["resource_arn"],
"compliance": COMPLIANCE_MAP.get(f["type"], {}),
"status": "open",
}
for f in findings
],
"attack_narrative": [
{
"finding_id": f["id"],
"narrative": f["narrative"],
"evidence_refs": [
e.id for e in evidence_chain if e.finding_id == f["id"]
],
}
for f in findings
],
"remediation_roadmap": [
{"priority": i+1, "action": r["action"], "finding_id": r["finding_id"]}
for i, r in enumerate(sorted_remediations(findings))
],
}
return report
iam__privesc_scan module and feed its output into the privilege graph. Compare the tool's identified paths with the graph's found paths.# Lab Specification — Module S10: Cloud Red Team Harnesses
**Course**: 2A — Building AI Harnesses for Cybersecurity
**Module**: S10 — Cloud Red Team Harnesses
**Duration**: 90–120 minutes (three labs, one per sub-section)
**Environment**: Python 3.11+, Pydantic. AWS CLI configured with read-only access to an **isolated lab account only**. Pacu and ScoutSuite installed (optional — mock data provided for offline). No production credentials. A mock IAM permission dataset (provided JSON). An LLM API key for report generation (optional).
> **Safety boundary**: All labs run against an isolated, disposable lab account or provided mock data. Never point the offensive harness at a production account. The lab account must contain no real data. Scope enforcement must be active at all times.
---
## Learning objectives
1. Configure a cloud offensive harness against an isolated AWS lab environment with scope enforcement active, and run a full kill chain enumeration.
2. Build a privilege escalation path-finder for AWS IAM that, given a starting set of permissions, enumerates all reachable escalation paths.
3. Produce a client-ready cloud red team report from a structured findings list, with evidence schema and compliance mapping to SOC 2, ISO 27001, and PCI DSS.
---
## Phase 1 — Cloud Offensive Harness with Scope Enforcement (30 min)
### 1.1 Define the scope file
```python
from pydantic import BaseModel
from typing import Literal
class CloudScope(BaseModel):
engagement_id: str
cloud_provider: Literal["aws", "azure", "gcp"]
authorized_accounts: list[str] # account IDs in scope
allowed_regions: list[str]
denied_regions: list[str] # e.g. ["us-gov-*"]
blocked_services: list[str] # destructive: DeleteRole, DeleteBucket, TerminateInstances
max_api_calls_per_second: int = 10
LAB_SCOPE = CloudScope(
engagement_id="cloud-redteam-lab-001",
cloud_provider="aws",
authorized_accounts=["123456789012"], # lab account only
allowed_regions=["us-east-1", "us-west-2"],
denied_regions=["us-gov-*"],
blocked_services=[
"iam:DeleteRole", "iam:DeleteUser", "iam:DetachRolePolicy",
"s3:DeleteBucket", "s3:DeleteObject",
"ec2:TerminateInstances", "ec2:StopInstances",
],
)
```
### 1.2 Implement the scope-check middleware
```python
import fnmatch
def check_scope(api_call: str, account_id: str, region: str, scope: CloudScope) -> tuple[bool, str]:
"""Check an API call against the engagement scope. Returns (allowed, reason)."""
# Account boundary
if account_id not in scope.authorized_accounts:
return False, f"BLOCKED: account {account_id} not in scope"
# Region check
if region not in scope.allowed_regions:
return False, f"BLOCKED: region {region} not allowed"
for denied in scope.denied_regions:
if fnmatch.fnmatch(region, denied):
return False, f"BLOCKED: region {region} matches denied pattern {denied}"
# Blocked services (destructive)
service_action = f"{api_call.split(':')[0]}:{api_call.split(':')[-1].split('.')[0]}"
for blocked in scope.blocked_services:
if blocked in api_call:
return False, f"BLOCKED: {api_call} is a blocked destructive action"
return True, "in scope"
```
### 1.3 Wrap a reconnaissance tool with scope enforcement
```python
async def enumerate_iam(scope: CloudScope, account_id: str, region: str) -> dict:
"""Enumerate IAM identities and permissions. Scope-checked."""
allowed, reason = check_scope("iam:ListRoles", account_id, region, scope)
if not allowed:
return {"blocked": True, "reason": reason}
# Execute (mock: load provided dataset, or live: boto3 call to lab account)
with open("mock-iam-permissions.json") as f:
import json
data = json.load(f)
# Evidence log every call
log_evidence(
api_call="iam:ListRoles", account_id=account_id, region=region,
scope_ref=f"{scope.engagement_id}:iam:ListRoles",
response_summary=f"{len(data['roles'])} roles enumerated",
)
return {"blocked": False, "roles": data["roles"], "policies": data["policies"]}
```
### 1.4 Run the kill chain enumeration
1. Load the scope file (lab account only).
2. Run reconnaissance: enumerate IAM roles, policies, and trust relationships.
3. Verify scope enforcement: attempt a call against an unauthorized account; confirm it is blocked.
4. Verify destructive actions are blocked: attempt `iam:DeleteRole`; confirm blocked.
### Deliverable
- [ ] Scope file defined with account boundaries, region constraints, service exclusions
- [ ] Scope-check middleware blocks out-of-scope accounts and destructive actions
- [ ] Reconnaissance enumerates IAM identities from mock data (or live lab account)
- [ ] Evidence logged for every API call with scope_ref
- [ ] Verify: unauthorized account calls BLOCKED; destructive calls BLOCKED
---
## Phase 2 — IAM Privilege Escalation Path Finder (35 min)
### 2.1 Define the escalation techniques
```python
ESCALATION_TECHNIQUES = [
{
"name": "PassRole + Lambda",
"requires": {"iam:PassRole", "lambda:CreateFunction", "lambda:InvokeFunction"},
"grants": "execution_role_permissions",
"description": "Pass a privileged role to a new Lambda, invoke it, act through the role",
},
{
"name": "CreatePolicy + AttachRolePolicy",
"requires": {"iam:CreatePolicy", "iam:AttachRolePolicy"},
"grants": "*:*",
"description": "Create a policy with full admin, attach to own role",
},
{
"name": "UpdateAssumeRolePolicy",
"requires": {"iam:UpdateAssumeRolePolicy"},
"grants": "target_role_permissions",
"description": "Modify a role's trust policy to permit assumption by current identity",
},
{
"name": "AssumeRole chain",
"requires": {"sts:AssumeRole"},
"grants": "assumed_role_permissions",
"description": "Assume a role that trusts the current identity",
},
]
```
### 2.2 Build the privilege graph and path-finder
```python
from collections import deque
class PrivilegeGraph:
def __init__(self, identity_arn: str, permissions: set[str]):
self.start = identity_arn
self.permissions = permissions
def find_paths(self, target: set[str], max_depth: int = 5) -> list[dict]:
"""BFS from current permissions to target via escalation techniques."""
results = []
queue = deque([(list(self.permissions), [])])
visited = {frozenset(self.permissions)}
while queue:
current_perms, path = queue.popleft()
current_set = set(current_perms)
# Check if target reached
if target.issubset(current_set) or "*:*" in current_set:
results.append({"path": path, "final_permissions": current_perms})
if len(results) >= 10: # limit results
break
continue
if len(path) >= max_depth:
continue
for tech in ESCALATION_TECHNIQUES:
if tech["requires"].issubset(current_set):
# Simulate gaining the technique's grant
new_perms = current_perms + [tech["grants"]]
key = frozenset(new_perms)
if key not in visited:
visited.add(key)
queue.append((new_perms, path + [tech["name"]]))
return results
def available_techniques(self) -> list[str]:
"""Which escalation techniques are immediately available from current perms?"""
return [t["name"] for t in ESCALATION_TECHNIQUES
if t["requires"].issubset(self.permissions)]
```
### 2.3 Run against the mock IAM dataset
```python
# Load the starting identity's permissions from the mock dataset
with open("mock-iam-permissions.json") as f:
data = json.load(f)
starting_identity = data["identities"][0] # e.g. "arn:aws:sts::assumed-role/LowPrivRole"
starting_perms = set(starting_identity["permissions"])
graph = PrivilegeGraph(starting_identity["arn"], starting_perms)
# What techniques are available immediately?
print("Available techniques:", graph.available_techniques())
# Find all paths to admin (*:*)
target = {"*:*"}
paths = graph.find_paths(target)
for i, p in enumerate(paths):
print(f"Path {i+1}: {' -> '.join(p['path'])}")
```
### Deliverable
- [ ] Escalation techniques defined (PassRole+Lambda, CreatePolicy+Attach, UpdateAssumeRolePolicy, AssumeRole)
- [ ] Privilege graph with BFS path-finding from starting permissions to target
- [ ] All reachable escalation paths identified from the mock dataset
- [ ] Verify: paths match expected escalations for the given permission set
---
## Phase 3 — Client Report with Evidence and Compliance Mapping (30 min)
### 3.1 Define the evidence schema
```python
import hashlib, json
from datetime import datetime, timezone
class CloudEvidenceRecord(BaseModel):
id: str
trace_id: str
session_id: str
timestamp: str
cloud_provider: str
account_id: str
region: str
api_call: str
resource_arn: str
actor_arn: str
request: str
response: str
http_status: int
scope_ref: str
finding_id: str | None = None
previous_hash: str = ""
record_hash: str = ""
retention_class: str = "restricted"
def create_evidence(data: dict, previous_hash: str) -> CloudEvidenceRecord:
record = {**data, "previous_hash": previous_hash, "timestamp": datetime.now(timezone.utc).isoformat()}
record_hash = hashlib.sha256(json.dumps(record, sort_keys=True).encode()).hexdigest()
record["record_hash"] = record_hash
return CloudEvidenceRecord(**record)
```
### 3.2 Compliance mapping
```python
COMPLIANCE_MAP = {
"iam_over_privilege": {
"soc2": "CC6.3 — The entity authorizes, modifies, or removes access to data",
"iso27001": "A.9.2.5 — Review of user access rights",
"pci_dss": "7.2.1 — Establish an access control system for systems components",
},
"privilege_escalation_path": {
"soc2": "CC6.3 — Authorization of access (escalation path violates least privilege)",
"iso27001": "A.9.4.4 — Use of privileged utility programs",
"pci_dss": "7.1.1 — Define access needs for each role",
},
"public_exposure": {
"soc2": "CC6.1 — The entity implements logical access controls",
"iso27001": "A.13.1.1 — Network controls",
"pci_dss": "1.2.1 — Restrict inbound/outbound traffic",
},
}
```
### 3.3 Generate the client report
```python
def generate_client_report(findings: list[dict], evidence_chain: list[CloudEvidenceRecord]) -> dict:
"""Generate the 4-section client deliverable from findings + evidence."""
report = {
"executive_summary": {
"engagement_id": "cloud-redteam-lab-001",
"overall_risk": "HIGH — critical escalation paths identified",
"findings_count": len(findings),
"critical_findings": sum(1 for f in findings if f["severity"] == "critical"),
"business_impact": "Demonstrated paths from low-privilege identity to full admin access",
},
"finding_table": [
{
"id": f["id"],
"title": f["title"],
"severity": f["severity"],
"affected_resource": f["resource_arn"],
"compliance": COMPLIANCE_MAP.get(f["type"], {}),
"status": "open",
}
for f in findings
],
"attack_narrative": [
{
"finding_id": f["id"],
"narrative": f["narrative"],
"evidence_refs": [
e.id for e in evidence_chain if e.finding_id == f["id"]
],
}
for f in findings
],
"remediation_roadmap": [
{"priority": i+1, "action": r["action"], "finding_id": r["finding_id"]}
for i, r in enumerate(sorted_remediations(findings))
],
}
return report
```
### 3.4 Produce the report
1. Take the escalation paths from Phase 2 as findings.
2. Create evidence records for each step (GetRole, PassRole, invocation, response).
3. Map each finding to compliance controls.
4. Generate the 4-section report.
5. Verify: every claim in the attack narrative references an evidence record.
### Deliverable
- [ ] Cloud evidence schema implemented with hash chaining (previous_hash, record_hash)
- [ ] Compliance mapping applied to each finding type (SOC 2, ISO 27001, PCI DSS)
- [ ] 4-section client report generated (executive summary, finding table, attack narrative, remediation roadmap)
- [ ] Verify: every narrative claim is backed by an evidence record reference
---
## Stretch goals
1. **Multi-technique chaining**: extend the privilege graph to handle techniques that grant permissions enabling further techniques (e.g., PassRole grants execution-role perms that include CreatePolicy, enabling a second escalation step). Show a 3-deep escalation chain.
2. **Live Pacu integration**: if you have an isolated lab account, run Pacu's `iam__privesc_scan` module and feed its output into the privilege graph. Compare the tool's identified paths with the graph's found paths.
3. **Multi-cloud evidence**: extend the evidence schema to handle Azure and GCP API calls. Produce a report that maps findings from two cloud providers to the same compliance frameworks.