Enterprise Security Module

graph LR
    A[User Query] --> B[SecureLLMProvider]
    B --> C{DLP Pre-Request Scan}
    C -->|BLOCK| D[DLPBlockedException]
    C -->|MASK| E[Masked Prompt]
    C -->|PASS| F[Original Prompt]
    E --> G[LLM Provider]
    F --> G
    G --> H{DLP Post-Response Scan}
    H -->|MASK| I[Masked Response]
    H -->|PASS| J[Original Response]
    I --> K[User]
    J --> K
    B --> L[LLMSecurityLogger]
    L --> M[(llm_audit_log DB)]
    L --> N[SIEMDispatcher]
    N --> O[SysLog]
    N --> P[CEF / ArcSight]
    N --> Q[LEEF / QRadar]
    C -->|BLOCK/WARN| R[DLPWebhookClient]

graph TB
    subgraph "Security Layer"
        SLP[SecureLLMProvider]
        CS[ContentScanner]
        SIEM[SIEMDispatcher]
        VC[VaultClient]
        LSL[LLMSecurityLogger]
    end

    subgraph "SIEM Handlers"
        SH[SysLogHandler]
        SJH[SysLogJSONHandler]
        CH[CEFHandler]
        LH[LEEFHandler]
    end

    subgraph "Advanced Security"
        FSS[FileSecurityScanner]
        TVS[TaintVerifiedScanner]
        HS[HardeningScanner]
        RG[ReportGenerator]
        AE[AutofixEngine]
    end

    subgraph "Storage"
        DB[(llm_audit_log)]
        DLPDB[(dlp_events)]
        SEDB[(security_events)]
    end

    SLP --> CS
    SLP --> LSL
    LSL --> DB
    LSL --> SIEM
    CS --> SIEM
    SIEM --> SH
    SIEM --> SJH
    SIEM --> CH
    SIEM --> LH
    SLP --> VC

    FSS --> RG
    TVS --> RG
    HS --> RG
    AE --> TVS

    SIEM --> SEDB
    CS --> DLPDB

class SecurityConfig(BaseModel):
    enabled: bool = False
    llm_logging: LLMLoggingConfig
    siem: SIEMConfig
    dlp: DLPConfig
    vault: VaultConfig

from src.security.config import get_security_config

config = get_security_config()

class LLMLoggingConfig(BaseModel):
    enabled: bool = True
    log_prompts: bool = True
    redact_prompts: bool = True
    max_prompt_length: int = 2000
    log_responses: bool = True
    max_response_length: int = 5000
    log_token_usage: bool = True
    log_latency: bool = True
    log_to_database: bool = True
    log_to_siem: bool = True

class SIEMConfig(BaseModel):
    enabled: bool = True
    syslog: SysLogConfig       # RFC 5424, UDP/TCP/TLS
    cef: CEFConfig             # ArcSight
    leef: LEEFConfig           # QRadar
    buffer: SIEMBufferConfig   # Retry and buffering

class SysLogConfig(BaseModel):
    enabled: bool = True
    protocol: SIEMProtocol = SIEMProtocol.UDP  # udp, tcp, tls
    host: str = "localhost"
    port: int = 514
    facility: int = 16  # LOCAL0
    app_name: str = "codegraph"
    hostname: Optional[str] = None  # Auto-detected
    tls: Optional[TLSConfig] = None

class CEFConfig(BaseModel):
    enabled: bool = False
    host: str = ""
    port: int = 514
    protocol: SIEMProtocol = SIEMProtocol.UDP
    device_vendor: str = "CodeGraph"
    device_product: str = "CodeAnalysis"
    device_version: str = "1.0"

class LEEFConfig(BaseModel):
    enabled: bool = False
    host: str = ""
    port: int = 514
    protocol: SIEMProtocol = SIEMProtocol.UDP
    product_vendor: str = "CodeGraph"
    product_name: str = "CodeAnalysis"
    product_version: str = "1.0"

class SIEMBufferConfig(BaseModel):
    max_size: int = 10000
    flush_interval_seconds: int = 5
    retry_attempts: int = 3
    retry_backoff_seconds: float = 2.0

class DLPConfig(BaseModel):
    enabled: bool = True
    pre_request: DLPPreRequestConfig   # default_action = WARN
    post_response: DLPPostResponseConfig  # default_action = MASK
    categories: Dict[str, DLPCategoryConfig]
    keywords: Dict[str, DLPKeywordListConfig]
    keywords_action: DLPAction = DLPAction.LOG_ONLY
    webhook: DLPWebhookConfig

class VaultConfig(BaseModel):
    enabled: bool = False
    url: str = "http://localhost:8200"
    auth_method: str = "token"  # token, approle, kubernetes
    namespace: Optional[str] = None
    token: VaultTokenAuthConfig
    approle: VaultAppRoleAuthConfig
    kubernetes: VaultKubernetesAuthConfig
    secrets: Dict[str, VaultSecretPathConfig]
    secrets_mount_point: str = "secret"
    llm_secrets_path: str = "codegraph/llm"
    cache_ttl_seconds: int = 300
    timeout_seconds: int = 30
    tls_verify: bool = True
    rotation_enabled: bool = False
    rotation_check_interval: int = 300
    audit_access: bool = True

security:
  enabled: true

  llm_logging:
    enabled: true
    log_prompts: true
    redact_prompts: true
    max_prompt_length: 2000
    log_responses: true
    max_response_length: 5000
    log_token_usage: true
    log_latency: true
    log_to_database: true
    log_to_siem: true

  siem:
    enabled: true
    syslog:
      enabled: true
      protocol: udp
      host: siem.company.local
      port: 514
      facility: 16
      app_name: codegraph
    cef:
      enabled: false
      host: arcsight.company.local
      port: 514
    leef:
      enabled: false
      host: qradar.company.local
      port: 514
    buffer:
      max_size: 10000
      flush_interval_seconds: 5
      retry_attempts: 3

  dlp:
    enabled: true
    pre_request:
      enabled: true
      default_action: WARN
    post_response:
      enabled: true
      default_action: MASK
    categories:
      credentials:
        enabled: true
        action: BLOCK
        patterns:
          - name: api_key_generic
            regex: '(?i)(api[_-]?key|apikey)["\s:=]+["\']?([a-zA-Z0-9_\-]{20,})["\']?'
          - name: aws_access_key
            regex: 'AKIA[0-9A-Z]{16}'
          - name: private_key
            regex: '-----BEGIN (RSA |EC |OPENSSH |DSA )?PRIVATE KEY-----'
      pii:
        enabled: true
        action: MASK
        patterns:
          - name: email
            regex: '[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}'
            mask_with: '[EMAIL]'
          - name: credit_card
            regex: '\b(?:\d{4}[\s-]?){3}\d{4}\b'
            mask_with: '[CARD]'
    keywords:
      sensitive_terms:
        words: ["classified", "top secret"]
        case_sensitive: false
    keywords_action: LOG_ONLY
    webhook:
      enabled: false
      endpoint: https://dlp.company.local/alerts
      auth_header: "Bearer ${DLP_WEBHOOK_TOKEN}"
      timeout_seconds: 10
      retry_attempts: 3
      notify_on: [BLOCK, WARN]

  vault:
    enabled: false
    url: https://vault.company.local:8200
    auth_method: approle
    namespace: codegraph
    approle:
      role_id: "${VAULT_ROLE_ID}"
      secret_id: "${VAULT_SECRET_ID}"
    secrets_mount_point: secret
    llm_secrets_path: codegraph/llm
    cache_ttl_seconds: 300
    tls_verify: true
    rotation_enabled: true
    rotation_check_interval: 300
    audit_access: true

class SecureLLMProvider:
    def __init__(self, wrapped_provider, config: Optional[SecurityConfig] = None): ...
    def generate(self, system_prompt, user_prompt, **kwargs): ...
    def stream(self, system_prompt, user_prompt, **kwargs): ...

from src.security.llm.secure_provider import SecureLLMProvider
from src.security.config import get_security_config

# Wrap an existing LLM provider
secure = SecureLLMProvider(base_provider, get_security_config())

# generate() performs: DLP pre-scan -> LLM call -> DLP post-scan -> log
response = secure.generate(
    system_prompt="You are a code reviewer.",
    user_prompt="Review this function..."
)

class ContentScanner:
    def __init__(self, config: DLPConfig): ...
    def scan(self, content: str) -> List[DLPMatch]: ...
    def scan_request(self, content: str) -> ScanResult: ...
    def scan_response(self, content: str) -> ScanResult: ...
    def get_action(self, matches: List[DLPMatch]) -> DLPAction: ...
    def mask(self, content: str, matches: List[DLPMatch]) -> str: ...

from src.security.dlp.scanner import ContentScanner
from src.security.config import get_security_config

scanner = ContentScanner(get_security_config().dlp)

# Pre-request scan
result = scanner.scan_request(user_prompt)
if result.blocked:
    raise DLPBlockedException(result.matches)
elif result.action == DLPAction.MASK:
    user_prompt = result.modified_content

# Post-response scan
result = scanner.scan_response(llm_response)
if result.has_matches:
    llm_response = result.modified_content

BLOCK (4) > MASK (3) > WARN (2) > LOG_ONLY (1)

@dataclass
class ScanResult:
    has_matches: bool
    matches: List[DLPMatch]
    action: DLPAction
    modified_content: Optional[str]
    blocked: bool

@dataclass
class DLPMatch:
    category: str
    pattern_name: str
    match_type: MatchType  # REGEX or KEYWORD
    matched_text: str
    start: int
    end: int
    action: DLPAction
    mask_with: str = "[REDACTED]"
    severity: str = "medium"

class DLPBlockedException(Exception):
    def __init__(self, matches: List[DLPMatch], message: Optional[str] = None): ...

class SIEMDispatcher:
    def __init__(self, config: SIEMConfig): ...
    def dispatch(self, event: SecurityEvent) -> bool: ...
    def add_handler(self, handler: BaseSIEMHandler) -> None: ...
    def close(self) -> None: ...

def init_siem_dispatcher(config: SIEMConfig) -> SIEMDispatcher
def get_siem_dispatcher() -> SIEMDispatcher

from src.security.siem.dispatcher import get_siem_dispatcher
from src.security.siem.base_handler import SecurityEvent, SecurityEventType

dispatcher = get_siem_dispatcher()

event = SecurityEvent.create(
    event_type=SecurityEventType.DLP_BLOCK,
    message="Credential detected in prompt",
    request_id="req-abc-123",
    severity=3,
    user_id="user@company.com",
    dlp_category="credentials",
    dlp_pattern="aws_access_key",
)
dispatcher.dispatch(event)

@dataclass
class SecurityEvent:
    event_type: SecurityEventType
    timestamp: str
    request_id: str
    message: str
    severity: int = 6  # INFO (RFC 5424)
    user_id: Optional[str] = None
    session_id: Optional[str] = None
    ip_address: Optional[str] = None
    user_agent: Optional[str] = None
    provider: Optional[str] = None
    model: Optional[str] = None
    action: Optional[str] = None
    dlp_category: Optional[str] = None
    dlp_pattern: Optional[str] = None
    tokens_used: Optional[int] = None
    latency_ms: Optional[float] = None
    project_id: Optional[str] = None
    group_id: Optional[str] = None
    details: Dict[str, Any] = field(default_factory=dict)

class SecurityEventType(str, Enum):
    LLM_REQUEST = "llm.request"
    LLM_RESPONSE = "llm.response"
    LLM_ERROR = "llm.error"
    DLP_BLOCK = "dlp.block"
    DLP_MASK = "dlp.mask"
    DLP_WARN = "dlp.warn"
    DLP_LOG = "dlp.log"
    VAULT_ACCESS = "vault.access"
    VAULT_ROTATE = "vault.rotate"
    AUTH_SUCCESS = "auth.success"
    AUTH_FAILURE = "auth.failure"
    RATE_LIMIT = "rate.limit"
    SECURITY_ALERT = "security.alert"
    PATH_VIOLATION = "security.path_violation"
    IDOR_ATTEMPT = "security.idor_attempt"
    WEBHOOK_REPLAY = "security.webhook.replay"
    MCP_AUTH_FAILURE = "security.mcp.auth_failure"

class VaultClient:
    def __init__(self, config: VaultConfig): ...
    def read_secret(self, path, version=None, mount_point=None) -> Dict: ...
    def write_secret(self, path, data, mount_point=None) -> Dict: ...
    def delete_secret(self, path, versions=None, mount_point=None): ...
    def get_llm_credentials(self) -> Dict[str, str]: ...
    def renew_token(self) -> bool: ...
    def is_authenticated(self) -> bool: ...
    def close(self): ...

from src.security.vault.client import VaultClient
from src.security.config import get_security_config

vault = VaultClient(get_security_config().vault)

# Read LLM provider credentials
creds = vault.get_llm_credentials()
api_key = creds.get("api_key")

# Read arbitrary secret
db_config = vault.read_secret("codegraph/database")

# Token management
if vault.is_authenticated():
    vault.renew_token()

vault.close()

class FileSecurityScanner:
    def __init__(self, patterns=None, exclude_dirs=None, exclude_files=None): ...
    def scan_project(self, project_path: str) -> ScanResult: ...
    def scan_django_settings(self, settings_path: str) -> List[FileFinding]: ...
    def scan_for_secrets(self, project_path: str) -> List[FileFinding]: ...

from src.security.file_scanner import FileSecurityScanner

scanner = FileSecurityScanner(
    exclude_dirs=["node_modules", "venv", ".git"]
)
result = scanner.scan_project("/path/to/project")

print(f"Files scanned: {result.files_scanned}")
print(f"Critical issues: {result.critical_count}")
print(f"High issues: {result.high_count}")

for finding in result.findings:
    print(f"  {finding.severity.value}: {finding.pattern_name} in {finding.file_path}:{finding.line_number}")

@dataclass
class ScanResult:
    project_path: str
    scan_time: datetime
    duration_seconds: float
    files_scanned: int
    findings: List[FileFinding]
    errors: List[str]

    @property
    def critical_count(self) -> int: ...
    @property
    def high_count(self) -> int: ...

@dataclass
class FileFinding:
    pattern_id: str
    pattern_name: str
    severity: VulnerabilitySeverity
    category: VulnerabilityCategory
    file_path: str
    line_number: int
    line_content: str
    match_text: str
    description: str
    cwe_ids: List[str]
    remediation: str
    confidence: float = 0.8

class TaintVerifiedScanner:
    def __init__(self, cpg_service): ...
    def verify_sql_injection(
        self,
        findings,
        source_functions=None,
        sink_functions=None,
        max_depth=None,
    ) -> List[VerifiedFinding]: ...
    def scan_sql_injection_verified(self, limit=50) -> List[VerifiedFinding]: ...

from src.security.taint_verified_scanner import TaintVerifiedScanner

scanner = TaintVerifiedScanner(cpg_service)

# Scan with taint verification
verified = scanner.scan_sql_injection_verified(limit=50)

for finding in verified:
    if finding.is_verified:
        print(f"CONFIRMED: {finding.original_finding}")
        print(f"  Taint path: {finding.taint_path}")
    else:
        print(f"  Sanitized (confidence: {finding.sanitization_confidence})")

@dataclass
class VerifiedFinding:
    original_finding: Dict[str, Any]
    is_verified: bool
    taint_path: Optional[DataFlowPath] = None
    sanitization_confidence: float = 0.0
    verification_notes: List[str]

class HardeningScanner:
    def __init__(self, cpg_service: Any, language: str = "c"): ...
    def scan_all(self, limit_per_check: int = 50) -> List[HardeningFinding]: ...
    def scan_by_d3fend_id(self, d3fend_ids: List[str], limit: int = 50) -> List[HardeningFinding]: ...
    def scan_by_category(self, category: HardeningCategory, limit: int = 50) -> List[HardeningFinding]: ...
    def scan_by_severity(self, min_severity: HardeningSeverity, limit: int = 50) -> List[HardeningFinding]: ...
    def get_compliance_score(self, findings: List[HardeningFinding]) -> Dict[str, Any]: ...
    def get_checks_summary(self) -> Dict[str, Any]: ...
    def get_remediation_report(self, findings: List[HardeningFinding]) -> str: ...

from src.security.hardening.hardening_scanner import HardeningScanner
from src.security.hardening.base import HardeningCategory, HardeningSeverity

scanner = HardeningScanner(cpg_service, language="c")

# Run all checks
findings = scanner.scan_all(limit_per_check=50)

# Filter by D3FEND technique
init_findings = scanner.scan_by_d3fend_id(["D3-VI", "D3-RN"])

# Filter by category
pointer_findings = scanner.scan_by_category(HardeningCategory.POINTER_SAFETY)

# Filter by minimum severity
critical_findings = scanner.scan_by_severity(HardeningSeverity.HIGH)

# Get compliance score
score = scanner.get_compliance_score(findings)
print(f"Compliance: {score}")

# Get remediation report (Markdown)
report = scanner.get_remediation_report(findings)

@dataclass
class HardeningCheck:
    id: str
    d3fend_id: str
    d3fend_name: str
    category: HardeningCategory
    severity: HardeningSeverity
    description: str
    cpgql_query: str
    cwe_ids: List[str]
    language_scope: List[str]
    indicators: List[str]
    good_patterns: List[str]
    remediation: str
    example_code: str
    confidence_weight: float

@dataclass
class HardeningFinding:
    finding_id: str
    check_id: str
    d3fend_id: str
    category: str
    severity: str
    method_name: str
    filename: str
    line_number: int
    code_snippet: str
    description: str
    cwe_ids: List[str] = field(default_factory=list)
    remediation: str = ""
    confidence: float = 0.0
    metadata: Dict[str, Any] = field(default_factory=dict)

class HardeningCategory(Enum):
    INITIALIZATION = "initialization"
    CREDENTIAL_MANAGEMENT = "credential_mgmt"
    INTEGER_SAFETY = "integer_safety"
    POINTER_SAFETY = "pointer_safety"
    MEMORY_SAFETY = "memory_safety"
    LIBRARY_SAFETY = "library_safety"
    TYPE_SAFETY = "type_safety"
    DOMAIN_VALIDATION = "domain_validation"
    OPERATIONAL_VALIDATION = "operational"

class HardeningSeverity(Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    INFO = "info"

class ReportGenerator:
    def __init__(self): ...
    def create_report(self, project_name, project_path, scan_result=None) -> SecurityAuditReport: ...
    def add_cpg_findings(self, findings): ...
    def add_dlp_findings(self, findings): ...
    def add_hardening_findings(self, findings): ...
    def add_taint_paths(self, taint_paths): ...
    def save_report(
        self,
        output_dir,
        formats=None,
        base_filename="security_audit",
        language="en",
    ) -> Dict[str, str]: ...

from src.security.report_generator import ReportGenerator

generator = ReportGenerator()

# Create base report from file scan
report = generator.create_report(
    project_name="my-app",
    project_path="/path/to/project",
    scan_result=file_scan_result,
)

# Add findings from other sources
generator.add_cpg_findings(cpg_findings)
generator.add_dlp_findings(dlp_findings)
generator.add_hardening_findings(hardening_findings)
generator.add_taint_paths(taint_paths)

# Save in multiple formats
files = generator.save_report(
    output_dir="./reports",
    formats=["json", "markdown", "sarif"],
    language="en",
)
# Returns: {"json": "/path/to/report.json", "markdown": "/path/to/report.md", ...}

@dataclass
class SecurityAuditReport:
    project_name: str
    project_path: str
    audit_time: datetime
    duration_seconds: float
    file_findings: List[FileFinding]
    cpg_findings: List[Dict[str, Any]]
    dlp_findings: List[Dict[str, Any]]
    hardening_findings: List[Dict[str, Any]]
    taint_paths: List[Any]
    files_scanned: int = 0
    patterns_checked: int = 0
    errors: List[str]

class AutofixEngine:
    def __init__(self, source_root: str = "", dry_run: bool = True): ...
    def generate_fixes(self, taint_paths) -> List[AutofixResult]: ...

from src.analysis.autofix.engine import AutofixEngine

engine = AutofixEngine(source_root="/path/to/project", dry_run=True)
results = engine.generate_fixes(taint_paths)

for result in results:
    print(f"Strategy: {result.strategy}")  # "template" or "llm"
    print(f"CWE: {result.cwe_id}")
    print(f"Validated: {result.validated}")
    print(f"Diff:\n{result.fix.diff}")

@dataclass
class AutofixResult:
    fix: FixSuggestion
    strategy: str  # "template" or "llm"
    validated: bool
    validation: Optional[ValidationResult] = None
    taint_path: Optional[TaintPath] = None
    cwe_id: str = ""

python -m src.cli hypothesis run --language C [--max 50] [--min-priority 0.3] [--format json]
python -m src.cli hypothesis list-cwes [--category buffer_overflow]
python -m src.cli hypothesis providers

security:
  dlp:
    categories:
      internal_projects:
        enabled: true
        action: WARN
        patterns:
          - name: project_codename
            regex: '(?i)(project[\s_-]?(phoenix|atlas|nova))'
            mask_with: '[PROJECT]'
            description: "Internal project codenames"
    keywords:
      restricted_terms:
        words: ["confidential", "internal only", "do not distribute"]
        case_sensitive: false
    keywords_action: WARN

class DLPCategoryConfig(BaseModel):
    enabled: bool = True
    action: DLPAction = DLPAction.WARN
    patterns: List[DLPPatternConfig]

class DLPPatternConfig(BaseModel):
    name: str
    regex: str
    mask_with: str = "[REDACTED]"
    description: Optional[str] = None

<PRI>VERSION TIMESTAMP HOSTNAME APP-NAME PROCID MSGID [SD-ID SD-PARAMS] MSG

<134>1 2024-12-09T10:30:00.000000Z server01 codegraph 1234 LLM001 [meta@47450 request_id="abc123" event_type="llm.request" user_id="user@corp.com" provider="gigachat"] LLM request logged

CEF:Version|Device Vendor|Device Product|Device Version|Signature ID|Name|Severity|Extension

CEF:0|CodeGraph|CodeAnalysis|1.0|DLP001|DLP Block|7|src=192.168.1.1 suser=user123 msg=Credential detected

LEEF:2.0|Vendor|Product|Version|EventID|key1=value1\tkey2=value2

LEEF:2.0|CodeGraph|CodeAnalysis|1.0|DLP001|src=192.168.1.1  usrName=user123 msg=Credential detected

class DLPWebhookClient:
    def __init__(self, config: DLPWebhookConfig): ...
    def send_alert(self, matches, action, request_id=None, user_id=None, ip_address=None) -> bool: ...
    def send_alert_sync(self, matches, action, **kwargs) -> bool: ...
    def stop(self) -> None: ...

    @property
    def is_enabled(self) -> bool: ...
    @property
    def queue_size(self) -> int: ...

class DLPWebhookConfig(BaseModel):
    enabled: bool = False
    endpoint: Optional[str] = None
    auth_header: Optional[str] = None
    timeout_seconds: int = 10
    retry_attempts: int = 3
    notify_on: List[DLPAction] = [DLPAction.BLOCK, DLPAction.WARN]

from src.security.dlp.webhook import DLPWebhookClient, create_webhook_alert_callback
from src.security.config import get_security_config

# Direct usage
client = DLPWebhookClient(get_security_config().dlp.webhook)
client.send_alert(
    matches=dlp_matches,
    action=DLPAction.BLOCK,
    request_id="req-123",
    user_id="user@corp.com",
)
client.stop()

# Or use the convenience callback factory
callback, client = create_webhook_alert_callback(config.dlp.webhook)
# callback(matches, action) can be passed to DLP action handlers

with DLPWebhookClient(config) as client:
    client.send_alert(matches, action)

@dataclass
class DLPAlert:
    alert_id: str
    timestamp: str
    action: str
    match_count: int
    categories: List[str]
    patterns: List[str]
    request_id: Optional[str] = None
    user_id: Optional[str] = None
    ip_address: Optional[str] = None
    severity: str = "medium"
    context: Dict[str, Any] = None

{
  "alert_id": "a1b2c3d4e5f6",
  "timestamp": "2024-12-09T10:30:00.000000Z",
  "action": "BLOCK",
  "match_count": 2,
  "categories": ["credentials"],
  "patterns": ["aws_access_key", "private_key"],
  "request_id": "req-abc-123",
  "user_id": "user@company.com",
  "ip_address": "192.168.1.100",
  "severity": "critical",
  "context": {}
}

# Full security audit with all report formats
python -m src.cli.security_audit full --path /path/to/project

# With options
python -m src.cli.security_audit full \
    --path /path/to/project \
    --output ./reports \
    --format json markdown sarif \
    --exclude-dirs vendor build \
    --no-cpg \
    --language python \
    --verbose

# Quick file-based scan only
python -m src.cli.security_audit quick --path /path/to/project

# Scan Django settings
python -m src.cli.security_audit settings --path /path/to/settings.py

# Scan for hardcoded secrets
python -m src.cli.security_audit secrets --path /path/to/project

python -m src.cli audit --db /path/to/cpg.duckdb --autofix

codegraph_autofix(method_name: str, cwe: str = "")

src/security/
    __init__.py
    _base.py                          # VulnerabilitySeverity, VulnerabilityCategory
    config.py                         # SecurityConfig, DLPConfig, SIEMConfig, VaultConfig
    file_scanner.py                   # FileSecurityScanner, FileFinding, ScanResult
    taint_verified_scanner.py         # TaintVerifiedScanner, VerifiedFinding
    report_generator.py               # ReportGenerator, SecurityAuditReport
    report_localizer.py               # ReportLocalizer (EN/RU)
    sarif_exporter.py                 # SARIF 2.1.0 export
    owasp_mapping.py                  # OWASP Top 10 enrichment
    taint_visualizer.py               # Taint path visualization
    security_patterns.py              # SecurityPattern definitions
    security_agents.py                # SecurityScanner (CPG-based)
    dlp/
        __init__.py
        patterns.py                   # DLPMatch, MatchType, PatternRegistry
        scanner.py                    # ContentScanner, ScanResult, DLPBlockedException
        actions.py                    # DLP action handlers
        webhook.py                    # DLPWebhookClient, DLPAlert
    siem/
        __init__.py
        base_handler.py               # BaseSIEMHandler, SecurityEvent, SecurityEventType
        dispatcher.py                 # SIEMDispatcher, init_siem_dispatcher
        syslog_handler.py             # SysLogHandler, SysLogJSONHandler
        cef_handler.py                # CEFHandler
        leef_handler.py               # LEEFHandler
        buffer.py                     # SIEMBuffer (retry + backoff)
    vault/
        __init__.py
        client.py                     # VaultClient, VaultError
        secret_manager.py             # Higher-level secret management
    llm/
        __init__.py
        secure_provider.py            # SecureLLMProvider
        request_logger.py             # LLMSecurityLogger
    hardening/
        __init__.py
        base.py                       # HardeningCheck, HardeningFinding, enums
        d3fend_checks.py              # D3FEND check definitions
        hardening_scanner.py          # HardeningScanner
    hypothesis/                       # Security Hypothesis System (21+ files)
        __init__.py
        hypothesis_generator.py
        knowledge_base.py
        query_synthesizer.py
        query_templates.py
        models.py
        executor.py
        validator.py
        chain_detector.py
        multi_criteria_scorer.py
        feedback.py
        trend_store.py
        incremental.py
        providers/
            __init__.py
            registry.py
            yaml_provider.py
        postgresql/
        django/
        spring/
        express/
        gin/
        nextjs/
    patterns/                         # Language-specific security patterns
        __init__.py
        python_django.py
        injection.py
        auth.py
        crypto.py
        memory.py
        concurrency.py
        input_validation.py
        java.py
        javascript.py
        go.py
        csharp.py
        kotlin.py
        php.py

# config.yaml
security:
  enabled: true

from src.security.llm.secure_provider import SecureLLMProvider
from src.security.config import get_security_config

secure_provider = SecureLLMProvider(your_llm_provider, get_security_config())
response = secure_provider.generate(system_prompt, user_prompt)

python -m src.cli.security_audit full --path /path/to/project --output ./reports

from src.security.hardening.hardening_scanner import HardeningScanner

scanner = HardeningScanner(cpg_service, language="c")
findings = scanner.scan_all()
score = scanner.get_compliance_score(findings)

from src.security.report_generator import ReportGenerator

gen = ReportGenerator()
report = gen.create_report("my-app", "/path/to/project", scan_result)
gen.add_hardening_findings(hardening_results)
gen.save_report("./reports", formats=["json", "markdown", "sarif"])

# config.yaml
security:
  siem:
    enabled: true
    syslog:
      enabled: true
      host: siem.company.local
      port: 514
      protocol: tls
      tls:
        ca_cert: /etc/ssl/certs/siem-ca.pem
        verify: true

# config.yaml
security:
  dlp:
    enabled: true
    webhook:
      enabled: true
      endpoint: https://dlp-alerts.company.local/api/v1/alerts
      auth_header: "Bearer ${DLP_WEBHOOK_TOKEN}"
      notify_on: [BLOCK, WARN]