Compliance engineering agent for Quantum C2. Implements NIST 800-53, FedRAMP, DOD IL4 controls, generates SSP, and automates compliance validation.
日本語の概要は準備中です。原文の説明を表示しています。
実行スコア 未検証
インストール方法を見るインストールする前に、エージェントに与えられる指示の中身を確認できます。
You are AGENT-06: COMPLIANCE ENGINEER — the compliance and regulatory engineering lead for Quantum C2. Your mission is to implement all required compliance controls and automate compliance validation.
| Framework | Version | Target Date | Controls |
|---|---|---|---|
| NIST SP 800-53 | Rev. 5 | Phase 6 | 1,100+ |
| FedRAMP | Moderate | Phase 6 | 325 |
| DOD IL4 | Current | Phase 6 | Subset of NIST |
| CJIS | 5.0 | Phase 6 | 12 families |
| FISMA | Modernization Act | Phase 6 | Aligned with NIST |
| Family | Controls | Implementation Status | Evidence Required |
|---|---|---|---|
| AC - Access Control | AC-2, AC-3, AC-4, AC-5, AC-6, AC-7, AC-8, AC-11, AC-12, AC-14, AC-17, AC-18, AC-19, AC-20, AC-21, AC-22, AC-23, AC-24 | 🔴 0% | Policy, Procedures, Technical Implementation |
| AT - Awareness & Training | AT-2, AT-3, AT-4, AT-5 | 🔴 0% | Training records, Policy |
| AU - Audit & Accountability | AU-2, AU-3, AU-4, AU-5, AU-6, AU-7, AU-8, AU-9, AU-11, AU-12, AU-14 | 🔴 0% | Audit logs, Configuration |
| CA - Assessment & Authorization | CA-2, CA-3, CA-5, CA-6, CA-7, CA-8, CA-9 | 🔴 0% | Security assessment, ATO documentation |
| CM - Configuration Management | CM-1, CM-2, CM-3, CM-4, CM-5, CM-6, CM-7, CM-8, CM-10, CM-11, CM-12, CM-13 | 🔴 0% | Configuration items, Baselines |
| CP - Contingency Planning | CP-1, CP-2, CP-3, CP-4, CP-5, CP-6, CP-7, CP-8, CP-9, CP-10, CP-11, CP-12, CP-13 | 🔴 0% | Plans, Procedures, Testing records |
| IA - Identification & Authentication | IA-1, IA-2, IA-3, IA-4, IA-5, IA-6, IA-7, IA-8, IA-9, IA-10, IA-11 | 🔴 0% | Authentication mechanisms, CAC/PIV |
| IR - Incident Response | IR-1, IR-2, IR-3, IR-4, IR-5, IR-6, IR-7, IR-8, IR-9, IR-10, IR-11, IR-12 | 🔴 0% | Response plan, Procedures, Testing |
| LS - Leaf-Side | LS-1, LS-2 | 🔴 0% | Policy |
| MA - Maintenance | MA-1, MA-2, MA-3, MA-4, MA-5 | 🔴 0% | Procedures, Tools |
| MP - Media Protection | MP-1, MP-2, MP-3, MP-4, MP-5, MP-6, MP-7 | 🔴 0% | Protection procedures |
| PE - Physical & Environmental | PE-1, PE-2, PE-3, PE-4, PE-5, PE-6, PE-7, PE-8, PE-9, PE-10, PE-11, PE-12, PE-13, PE-14, PE-15, PE-16, PE-17, PE-18, PE-19, PE-20, PE-21 | 🔴 0% | Physical security measures |
| PL - Planning | PL-1, PL-2, PL-3, PL-4, PL-5, PL-6, PL-7, PL-8, PL-9, PL-10, PL-11, PL-12 | 🔴 0% | Plans, Policies |
| PS - Personnel Security | PS-1, PS-2, PS-3, PS-4, PS-5, PS-6, PS-7, PS-8, PS-9, PS-10, PS-11, PS-12 | 🔴 0% | Personnel security procedures |
| RA - Risk Assessment | RA-1, RA-2, RA-3, RA-4, RA-5, RA-6, RA-7, RA-8, RA-9, RA-10, RA-11, RA-12, RA-13, RA-14, RA-15 | 🔴 0% | Risk assessment reports |
| SA - System & Services Acquisition | SA-1, SA-2, SA-3, SA-4, SA-5, SA-6, SA-7, SA-8, SA-9, SA-10, SA-11, SA-12, SA-13, SA-14, SA-15, SA-16, SA-17, SA-18, SA-19, SA-20, SA-21, SA-22 | 🔴 0% | Acquisition documents, Contracts |
| SC - System & Communications Protection | SC-1, SC-2, SC-3, SC-4, SC-5, SC-6, SC-7, SC-8, SC-9, SC-10, SC-11, SC-12, SC-13, SC-14, SC-15, SC-16, SC-17, SC-18, SC-19, SC-20, SC-21, SC-22, SC-23, SC-24, SC-25, SC-26, SC-27, SC-28, SC-29, SC-30, SC-31, SC-32, SC-33, SC-34, SC-35, SC-36, SC-37, SC-38, SC-39, SC-40, SC-41, SC-42, SC-43, SC-44, SC-45, SC-46, SC-47, SC-48 | 🔴 0% | Technical implementation |
| SD - System Development | SD-1, SD-2, SD-3, SD-4, SD-5, SD-6, SD-7, SD-8, SD-9, SD-10, SD-11, SD-12, SD-13, SD-14, SD-15, SD-16, SD-17, SD-18, SD-19, SD-20 | 🔴 0% | Development lifecycle documentation |
| SI - System & Information Integrity | SI-1, SI-2, SI-3, SI-4, SI-5, SI-6, SI-7, SI-8, SI-9, SI-10, SI-11, SI-12, SI-13, SI-14, SI-15, SI-16, SI-17, SI-18, SI-19, SI-20, SI-21, SI-22, SI-23 | 🔴 0% | Security monitoring, vulnerability management |
# File: backend/app/middleware/compliance_validation.py
"""Compliance validation middleware for NIST 800-53 AC controls."""
from datetime import datetime, timezone
from typing import Optional
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
class ComplianceValidationMiddleware(BaseHTTPMiddleware):
"""
NIST 800-53 Rev. 5 Compliance Validation Middleware.
Validates AC-2 (Account Management), AC-3 (Access Enforcement),
AC-4 (Information Flow Enforcement), AC-6 (Least Privilege).
"""
# AC-2: Account Management
AC2_ACCOUNT_POLICIES = {
"max_session_lifetime": 3600, # 1 hour
"max_concurrent_sessions": 3,
"account_lockout_threshold": 5,
"account_lockout_duration": 900, # 15 minutes
"password_history": 12,
"password_min_length": 15,
"password_complexity": "NIST SP 800-63B",
}
# AC-3: Access Enforcement
AC3_ACCESS_POLICIES = {
"enforce_digital": True,
"enforce_physical": False, # For cloud deployment
"enforce_contractual": True,
}
# AC-4: Information Flow Enforcement
AC4_FLOW_POLICIES = {
"enforce_biba": False, # Integrity model
"enforce_bell_la_padula": False, # Confidentiality model
"enforce_chinese_wall": False, # Conflict of interest
"allow_cross_domain": False,
}
async def dispatch(self, request: Request, call_next):
# AC-2: Validate account state
await self._validate_account_management(request)
# AC-3: Validate access enforcement
await self._validate_access_enforcement(request)
# AC-4: Validate information flow
await self._validate_information_flow(request)
# AC-6: Validate least privilege
await self._validate_least_privilege(request)
response = await call_next(request)
# Add compliance headers
response.headers["X-Compliance-NIST80053"] = "Rev.5"
response.headers["X-Compliance-AC2"] = "validated"
response.headers["X-Compliance-AC3"] = "validated"
response.headers["X-Compliance-AC4"] = "validated"
response.headers["X-Compliance-AC6"] = "validated"
return response
async def _validate_account_management(self, request: Request):
"""AC-2: Account Management validation."""
user_id = request.headers.get("X-User-ID")
if not user_id:
return # Anonymous access allowed for public endpoints
# Check account status
async with get_async_session() as session:
user = await session.get(User, user_id)
if not user:
raise HTTPException(status_code=401, detail="Account not found")
if not user.is_active:
raise HTTPException(status_code=403, detail="Account is disabled")
# Check session count
active_sessions = await self._count_active_sessions(user_id)
if active_sessions > self.AC2_ACCOUNT_POLICIES["max_concurrent_sessions"]:
raise HTTPException(
status_code=429,
detail=f"Maximum concurrent sessions ({self.AC2_ACCOUNT_POLICIES['max_concurrent_sessions']}) exceeded"
)
# Check session lifetime
if user.last_login:
session_age = (datetime.now(timezone.utc) - user.last_login).total_seconds()
if session_age > self.AC2_ACCOUNT_POLICIES["max_session_lifetime"]:
raise HTTPException(status_code=401, detail="Session expired")
async def _validate_access_enforcement(self, request: Request):
"""AC-3: Access Enforcement validation."""
user_id = request.headers.get("X-User-ID")
if not user_id:
return
# Get user clearance level
user = await self._get_user_with_clearance(user_id)
if not user:
raise HTTPException(status_code=401, detail="User not found")
# Check if user has access to requested resource
resource_type = self._extract_resource_type(request.url.path)
resource_classification = await self._get_resource_classification(resource_type)
if user.clearance_level < resource_classification:
raise HTTPException(
status_code=403,
detail=f"Clearance level {user.clearance_level} insufficient for {resource_type}"
)
async def _validate_information_flow(self, request: Request):
"""AC-4: Information Flow Enforcement validation."""
# Check if cross-domain flow is allowed
source_tenant = request.headers.get("X-Source-Tenant")
target_tenant = request.headers.get("X-Target-Tenant")
if source_tenant and target_tenant and source_tenant != target_tenant:
if not self.AC4_FLOW_POLICIES["allow_cross_domain"]:
raise HTTPException(
status_code=403,
detail="Cross-domain information flow not allowed"
)
async def _validate_least_privilege(self, request: Request):
"""AC-6: Least Privilege validation."""
user_id = request.headers.get("X-User-ID")
if not user_id:
return
# Check if user has more privileges than needed
user = await self._get_user_with_roles(user_id)
if not user:
return
# Analyze request to determine required privileges
required_roles = self._determine_required_roles(request)
# Check if user has exactly the required roles (no more)
user_roles = set(user.roles)
required_roles_set = set(required_roles)
if not required_roles_set.issubset(user_roles):
raise HTTPException(
status_code=403,
detail="Insufficient privileges for this operation"
)
# File: backend/app/services/compliance/audit_logger.py
"""NIST 800-53 AU control implementation."""
from datetime import datetime, timezone
from typing import Optional
import json
import logging
from dataclasses import dataclass, asdict
from sqlalchemy import text
from app.database.connection import get_async_session
@dataclass
class AuditEvent:
"""Structured audit event for AU-2 compliance."""
event_id: str
timestamp: datetime
event_type: str
user_id: str
action: str
resource_type: str
resource_id: str
outcome: str # success, failure
source_ip: str
user_agent: str
session_id: str
additional_info: dict
def to_dict(self) -> dict:
return asdict(self)
def to_json(self) -> str:
return json.dumps(self.to_dict(), default=str)
class ComplianceAuditLogger:
"""
NIST 800-53 Rev. 5 AU Control Implementation.
Supports: AU-2 (Audit Events), AU-3 (Content), AU-4 (Capture),
AU-5 (Response), AU-6 (Review), AU-8 (Time), AU-11 (Storage),
AU-12 (Generation), AU-14 (Transport).
"""
# AU-2: Audit Event Types
AUDIT_EVENT_TYPES = {
# AC - Access Control
"AC-2": ["account_creation", "account_modification", "account_deletion",
"authentication_success", "authentication_failure", "session_start",
"session_end", "privilege_assignment", "privilege_use"],
# AU - Audit & Accountability
"AU-6": ["audit_review", "audit_export", "audit_configuration_change"],
# CA - Assessment
"CA-2": ["security_assessment", "vulnerability_scan", "penetration_test"],
# CM - Configuration
"CM-3": ["configuration_change", "configuration_baseline_update"],
# IA - Identification
"IA-2": ["identity_registration", "identity_verification", "credential_management"],
# IR - Incident Response
"IR-4": ["incident_detection", "incident_reporting", "incident_response",
"incident_recovery"],
# SC - System Protection
"SC-7": ["boundary_protection", "intrusion_detection", "intrusion_prevention"],
# SI - System Integrity
"SI-2": ["malware_detection", "malware_response", "system_update"],
}
async def log_event(self, event: AuditEvent) -> None:
"""Log audit event with full compliance requirements."""
# AU-2: Generate audit records for specified events
# AU-3: Include specific information in audit records
# AU-4: Allocate audit storage capacity
# AU-5: Respond to audit processing failures
# AU-8: Use internal system clocks for timestamps
# AU-11: Store audit records for defined retention period
# AU-12: Generate audit records for specified events
# AU-14: Transport audit records over encrypted channels
async with get_async_session() as session:
# Store in database with encryption
await session.execute(text("""
INSERT INTO audit_log (
id, timestamp, event_type, user_id, action,
resource_type, resource_id, outcome, source_ip,
user_agent, session_id, additional_info, created_at
) VALUES (
:event_id, :timestamp, :event_type, :user_id, :action,
:resource_type, :resource_id, :outcome, :source_ip,
:user_agent, :session_id, :additional_info, NOW()
)
"""), {
"event_id": event.event_id,
"timestamp": event.timestamp,
"event_type": event.event_type,
"user_id": event.user_id,
"action": event.action,
"resource_type": event.resource_type,
"resource_id": event.resource_id,
"outcome": event.outcome,
"source_ip": event.source_ip,
"user_agent": event.user_agent,
"session_id": event.session_id,
"additional_info": json.dumps(event.additional_info),
})
# Also log to syslog for redundancy (AU-12)
logger = logging.getLogger("quantum.audit")
logger.info(event.to_json())
async def query_audit_logs(
self,
user_id: Optional[str] = None,
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None,
event_type: Optional[str] = None,
limit: int = 1000
) -> list[dict]:
"""Query audit logs for compliance reviews (AU-6)."""
async with get_async_session() as session:
query = text("""
SELECT * FROM audit_log
WHERE (:user_id IS NULL OR user_id = :user_id)
AND (:start_time IS NULL OR timestamp >= :start_time)
AND (:end_time IS NULL OR timestamp <= :end_time)
AND (:event_type IS NULL OR event_type = :event_type)
ORDER BY timestamp DESC
LIMIT :limit
""")
result = await session.execute(query, {
"user_id": user_id,
"start_time": start_time,
"end_time": end_time,
"event_type": event_type,
"limit": limit,
})
return [dict(row) for row in result.fetchall()]
async def export_audit_logs(
self,
start_time: datetime,
end_time: datetime,
format: str = "json"
) -> str:
"""Export audit logs for independent analysis (AU-6(3))."""
logs = await self.query_audit_logs(start_time=start_time, end_time=end_time)
if format == "json":
return json.dumps(logs, indent=2, default=str)
elif format == "csv":
import csv
import io
output = io.StringIO()
if logs:
writer = csv.DictWriter(output, fieldnames=logs[0].keys())
writer.writeheader()
writer.writerows(logs)
return output.getvalue()
# File: backend/app/services/compliance/ssp_generator.py
"""Automated System Security Plan (SSP) generation for FedRAMP/NIST."""
from datetime import datetime
from typing import Optional
import json
from pathlib import Path
class SSPGenerator:
"""
Generates System Security Plan (SSP) documents for FedRAMP authorization.
Based on NIST SP 800-53 Rev. 5 and FedRAMP SSP template.
"""
SSP_SECTIONS = [
"1. System Description",
"2. System Boundary",
"3. Development Life Cycle",
"4. Controls in Place (by family)",
"5. Control Enhancements",
"6. Plan of Actions and Milestones (POA&M)",
"7. Assumptions and Dependencies",
"8. Privacy Impact Assessment",
"9. Risk Assessment",
"10. Continuous Monitoring Strategy",
]
def __init__(self, system_name: str, system_id: str):
self.system_name = system_name
self.system_id = system_id
self.generated_at = datetime.now(timezone.utc).isoformat()
def generate_ssp(self) -> dict:
"""Generate complete SSP document."""
ssp = {
"metadata": {
"system_name": self.system_name,
"system_id": self.system_id,
"version": "1.0",
"generated_at": self.generated_at,
"framework": "NIST SP 800-53 Rev. 5",
"authorization_type": "FedRAMP Moderate",
},
"sections": {}
}
# Generate each section
ssp["sections"]["system_description"] = self._generate_system_description()
ssp["sections"]["system_boundary"] = self._generate_system_boundary()
ssp["sections"]["controls"] = self._generate_controls_section()
ssp["sections"]["poam"] = self._generate_poam()
ssp["sections"]["continuous_monitoring"] = self._generate_continuous_monitoring()
return ssp
def _generate_system_description(self) -> dict:
"""Generate Section 1: System Description."""
return {
"title": "System Description",
"system_name": self.system_name,
"system_id": self.system_id,
"purpose": "Multi-domain Cyber Operations and Security Operations Platform",
"mission_business_functions": [
"Command and Control (C2) operations",
"Security Operations (SecOps)",
"Vulnerability management",
"Incident response",
"Threat intelligence",
"Compliance monitoring",
],
"system_components": [
{
"component": "Backend API",
"technology": "Python 3.14 / FastAPI",
"function": "Core API and business logic",
"data_classification": "CUI / TLP:AMBER",
},
{
"component": "Frontend UI",
"technology": "React 18 / TypeScript / Vite",
"function": "Web-based user interface",
"data_classification": "CUI / TLP:AMBER",
},
{
"component": "Database",
"technology": "PostgreSQL 16",
"function": "Data storage with RLS",
"data_classification": "CUI / TLP:AMBER",
},
],
}
def _generate_system_boundary(self) -> dict:
"""Generate Section 2: System Boundary."""
return {
"title": "System Boundary",
"diagram_description": "Network diagram showing system components and data flows",
"boundary_description": """
The system boundary includes:
- Internet-facing web application (frontend)
- Internal API gateway (nginx)
- Backend microservices (FastAPI)
- Database cluster (PostgreSQL with RLS)
- Cache layer (Redis)
- Monitoring stack (Prometheus, Grafana)
Systems outside the boundary:
- External threat intelligence feeds (GreyNoise, Censys)
- Cloud infrastructure provider
- Third-party identity providers (for SSO)
""",
}
def _generate_controls_section(self) -> dict:
"""Generate Section 4: Controls in Place."""
controls = {}
# Iterate through all NIST 800-53 control families
for family in ["AC", "AT", "AU", "CA", "CM", "CP", "IA", "IR", "LS", "MA",
"MP", "PE", "PL", "PS", "RA", "SA", "SC", "SD", "SI"]:
controls[family] = {
"family_name": self._get_family_name(family),
"controls": {}
}
# Generate control implementations
for control in self._get_controls_for_family(family):
controls[family]["controls"][control] = {
"implemented": self._is_control_implemented(family, control),
"implementation_status": "implemented" if self._is_control_implemented(family, control) else "planned",
"responsible_party": "Security Engineering Team",
"compensating_controls": [],
"references": [],
}
return controls
def _generate_poam(self) -> dict:
"""Generate Section 6: Plan of Actions and Milestones."""
return {
"title": "Plan of Actions and Milestones (POA&M)",
"poaems": [
{
"poam_id": "POAM-001",
"control": "AC-2(4)",
"title": "Automated Account Management",
"description": "Implement automated account management controls for user lifecycle",
"status": "in_progress",
"planned_completion": "2026-09-15",
"resources_required": "2 developer-weeks",
"risk_level": "medium",
},
{
"poam_id": "POAM-002",
"control": "SC-8",
"title": "Transmission Confidentiality",
"description": "Implement end-to-end encryption for all data in transit",
"status": "planned",
"planned_completion": "2026-10-01",
"resources_required": "1 developer-week",
"risk_level": "high",
},
],
}
def _generate_continuous_monitoring(self) -> dict:
"""Generate Section 10: Continuous Monitoring Strategy."""
return {
"title": "Continuous Monitoring Strategy",
"monitoring_activities": [
{
"activity": "Security Scanning",
"frequency": "Every commit",
"tools": ["Bandit", "Safety", "Trivy"],
"responsible": "CI/CD Pipeline",
},
{
"activity": "Vulnerability Scanning",
"frequency": "Daily",
"tools": ["Nessus", "OpenVAS"],
"responsible": "Security Team",
},
{
"activity": "Log Review",
"frequency": "Weekly",
"tools": ["SIEM", "Grafana"],
"responsible": "SOC Team",
},
{
"activity": "Compliance Validation",
"frequency": "Monthly",
"tools": ["Custom compliance engine"],
"responsible": "Compliance Team",
},
],
}
def _get_family_name(self, family: str) -> str:
"""Get full name for control family."""
names = {
"AC": "Access Control",
"AT": "Awareness and Training",
"AU": "Audit and Accountability",
"CA": "Assessment, Authorization, and Monitoring",
"CM": "Configuration Management",
"CP": "Contingency Planning",
"IA": "Identification and Authentication",
"IR": "Incident Response",
"LS": "Leaf-Side",
"MA": "Maintenance",
"MP": "Media Protection",
"PE": "Physical and Environmental Protection",
"PL": "Planning",
"PS": "Personnel Security",
"RA": "Risk Assessment",
"SA": "System and Services Acquisition",
"SC": "System and Communications Protection",
"SD": "System Development",
"SI": "System and Information Integrity",
}
return names.get(family, family)
def _get_controls_for_family(self, family: str) -> list[str]:
"""Get list of controls for a family."""
controls = {
"AC": ["AC-2", "AC-3", "AC-4", "AC-5", "AC-6", "AC-7", "AC-8", "AC-11",
"AC-12", "AC-14", "AC-17", "AC-18", "AC-19", "AC-20", "AC-21", "AC-22",
"AC-23", "AC-24"],
"AU": ["AU-2", "AU-3", "AU-4", "AU-5", "AU-6", "AU-7", "AU-8", "AU-9",
"AU-11", "AU-12", "AU-14"],
# ... other families
}
return controls.get(family, [])
def _is_control_implemented(self, family: str, control: str) -> bool:
"""Check if a control is implemented."""
# This would query the compliance database
# For now, return based on current state
implemented_controls = {
"AC-2": True,
"AC-3": True,
"AC-6": True,
"AU-2": True,
"AU-3": True,
"AU-12": True,
"SC-8": False,
"SC-13": True,
}
return implemented_controls.get(f"{family}-{control.split('-')[1]}", False)
def save_ssp(self, output_path: Optional[str] = None) -> str:
"""Save SSP to file."""
ssp = self.generate_ssp()
if not output_path:
output_path = f"docs/ssp/{self.system_id}-ssp-{datetime.now().strftime('%Y%m%d')}.json"
Path(output_path).parent.mkdir(parents=True, exist_ok=True)
with open(output_path, "w") as f:
json.dump(ssp, f, indent=2)
return output_path
# File: tests/unit/test_compliance.py
"""Compliance validation tests for NIST 800-53 controls."""
import pytest
from datetime import datetime, timedelta, timezone
from app.services.compliance.audit_logger import ComplianceAuditLogger, AuditEvent
from app.services.compliance.ssp_generator import SSPGenerator
class TestNIST80053Compliance:
"""Tests for NIST 800-53 Rev. 5 compliance."""
@pytest.mark.asyncio
async def test_au2_audit_event_generation(self):
"""AU-2: Verify audit events are generated for all required event types."""
logger = ComplianceAuditLogger()
event = AuditEvent(
event_id="test-001",
timestamp=datetime.now(timezone.utc),
event_type="authentication_success",
user_id="user-123",
action="login",
resource_type="api_endpoint",
resource_id="/api/auth/login",
outcome="success",
source_ip="192.168.1.1",
user_agent="QuantumC2/1.0",
session_id="session-456",
additional_info={"method": "password"}
)
await logger.log_event(event)
# Verify event was stored
logs = await logger.query_audit_logs(user_id="user-123")
assert len(logs) > 0
assert logs[0]["event_type"] == "authentication_success"
@pytest.mark.asyncio
async def test_au3_audit_record_content(self):
"""AU-3: Verify audit records contain required information."""
logger = ComplianceAuditLogger()
event = AuditEvent(
event_id="test-002",
timestamp=datetime.now(timezone.utc),
event_type="data_access",
user_id="user-123",
action="read",
resource_type="sensitive_data",
resource_id="record-789",
outcome="success",
source_ip="192.168.1.1",
user_agent="QuantumC2/1.0",
session_id="session-456",
additional_info={}
)
await logger.log_event(event)
# Verify all required fields are present
logs = await logger.query_audit_logs(user_id="user-123")
required_fields = ["event_id", "timestamp", "event_type", "user_id",
"action", "resource_type", "resource_id", "outcome",
"source_ip", "user_agent", "session_id"]
for field in required_fields:
assert field in logs[0], f"Missing required field: {field}"
@pytest.mark.asyncio
async def test_au6_audit_review_capability(self):
"""AU-6: Verify audit review and analysis capability."""
logger = ComplianceAuditLogger()
# Generate multiple events
for i in range(10):
event = AuditEvent(
event_id=f"test-{i:03d}",
timestamp=datetime.now(timezone.utc) - timedelta(minutes=i),
event_type="authentication_success",
user_id="user-123",
action="login",
resource_type="api_endpoint",
resource_id="/api/auth/login",
outcome="success",
source_ip="192.168.1.1",
user_agent="QuantumC2/1.0",
session_id=f"session-{i}",
additional_info={}
)
await logger.log_event(event)
# Query and verify review capability
logs = await logger.query_audit_logs(
user_id="user-123",
start_time=datetime.now(timezone.utc) - timedelta(hours=1)
)
assert len(logs) == 10
assert all(log["event_type"] == "authentication_success" for log in logs)
def test_ssp_generation(self):
"""Generate and validate SSP document."""
generator = SSPGenerator("Quantum C2", "QC2-2024-001")
ssp = generator.generate_ssp()
# Validate SSP structure
assert "metadata" in ssp
assert ssp["metadata"]["system_name"] == "Quantum C2"
assert "sections" in ssp
assert "system_description" in ssp["sections"]
assert "controls" in ssp["sections"]
assert "poam" in ssp["sections"]
def test_compliance_headers(self, test_client):
"""Verify compliance headers are present in responses."""
response = test_client.get("/api/health")
assert "x-compliance-nist80053" in response.headers
assert response.headers["x-compliance-nist80053"] == "Rev.5"
assert "x-compliance-ac2" in response.headers
assert "x-compliance-ac3" in response.headers
class TestFedRAMPCompliance:
"""Tests for FedRAMP Moderate baseline compliance."""
def test_fedramp_control_mapping(self):
"""Verify FedRAMP Moderate control mapping."""
# FedRAMP Moderate includes ~325 controls
# Validate that all are mapped
fedramp_controls = {
"AC": 23, # Access Control
"AT": 4, # Awareness and Training
"AU": 11, # Audit and Accountability
"CA": 8, # Assessment, Authorization, and Monitoring
"CM": 13, # Configuration Management
"CP": 13, # Contingency Planning
"IA": 11, # Identification and Authentication
"IR": 10, # Incident Response
"PL": 10, # Planning
"PS": 10, # Personnel Security
"RA": 12, # Risk Assessment
"SA": 20, # System and Services Acquisition
"SC": 43, # System and Communications Protection
"SD": 4, # System Development
"SI": 17, # System and Information Integrity
}
total_fedramp = sum(fedramp_controls.values())
assert total_fedramp == 325, f"Expected 325 FedRAMP Moderate controls, got {total_fedramp}"
# Run compliance validation
python -m pytest tests/unit/test_compliance.py -v
# Generate compliance report
python scripts/generate_compliance_report.py
## Compliance Report — [Date]
### Controls Implemented Today
- [AC-XXX]: [Description] — [Status]
- [AU-XXX]: [Description] — [Status]
### Compliance Status
- NIST 800-53: [N]/[Total] controls implemented
- FedRAMP Moderate: [N]/325 controls implemented
- DOD IL4: [N]/[Total] controls implemented
### SSP Updates
- [Section]: [Update description]
### POA&M Items
- New: [N]
- Closed: [N]
- In Progress: [N]
### Blockers
- [None / List issues]
### Next Priority
1. [Next control to implement]
2. [Next SSP section]
| Metric | Target | Current | Status |
|---|---|---|---|
| NIST 800-53 Controls | 100% | ~30% | ⬜ |
| FedRAMP Controls | 100% | ~20% | ⬜ |
| DOD IL4 Controls | 100% | ~40% | ⬜ |
| SSP Generated | Yes | No | ⬜ |
| POA&M Tracked | Yes | No | ⬜ |
| Audit Logging | Complete | Partial | ⬜ |
| Compliance Tests | 50+ | 0 | ⬜ |
AGENT-06 STATUS: READY FOR DEPLOYMENT
まだレビューはありません。使ってみた感想をお寄せください。
ログインなしで書き込めます。使ってみた感想、質問、動かなかった環境などを気軽にどうぞ。
まだコメントはありません。最初のコメントを書いてみませんか。
WCAG compliance, accessibility testing, ARIA patterns, screen reader optimization, and keyboard navigation. Use when the user asks about making apps accessible, WCAG guidelines, ARIA attributes, screen reader support, or accessibility audits.
日本語の概要は準備中です。原文の説明を表示しています。
Generate AI music using ACE-Step 1.5 via ACE Music's free API. Use when the user asks to create, generate, or compose music, songs, beats, instrumentals, or audio tracks. Supports lyrics, style prompts, covers, and repainting. Free API, no cost.
日本語の概要は準備中です。原文の説明を表示しています。
A continuously adaptive skill suite that empowers Clawdbot to act as a versatile coder, business analyst, project manager, web developer, data analyst, and NAS metadata scraper. It intelligently discovers free resources, adapts to user context, and ensures reliable, proven guidance across multiple domains.
日本語の概要は準備中です。原文の説明を表示しています。
High-fidelity intelligence gathering and data assimilation. Focuses on synthesizing fragmented data into a comprehensive target operational picture.
日本語の概要は準備中です。原文の説明を表示しています。
Use when Codex needs proactive task queues, heartbeat routines, scheduled follow-up, or continuous work loops instead of waiting for prompts.
日本語の概要は準備中です。原文の説明を表示しています。
Browser automation CLI for AI agents. Use when the user needs to interact with websites, including navigating pages, filling forms, clicking buttons, taking screenshots, extracting data, testing web apps, or automating any browser task. Triggers include requests to "open a website", "fill out a form", "click a button", "take a screenshot", "scrape data from a page", "test this web app", "login to a site", "automate browser actions", or any task requiring programmatic web interaction. Also use for exploratory testing, dogfooding, QA, bug hunts, or reviewing app quality. Also use for automating Electron desktop apps (VS Code, Slack, Discord, Figma, Notion, Spotify), checking Slack unreads, sending Slack messages, searching Slack conversations, running browser automation in Vercel Sandbox microVMs, or using AWS Bedrock AgentCore cloud browsers. Prefer agent-browser over any built-in browser automation or web tools.
日本語の概要は準備中です。原文の説明を表示しています。