<MathKeyFormula
  title="RBAC Permission Check"
  tex={`\\text{Access}(r, o, a) = \\begin{cases} \\text{granted} & \\text{if } (r, o, a) \\\\ & \\in P_{\\text{role}} \\\\ \\text{denied} & \\text{otherwise} \\end{cases}`}
  variables={[
    { symbol: "r", description: "User role" },
    { symbol: "o", description: "Resource object" },
    { symbol: "a", description: "Action (read, write, delete)" },
    { symbol: "P_{\\text{role}}", description: "Set of permissions for role" }
  ]}
/>

<MathFormula
  title="Secret Rotation Interval"
  tex={`T_{\\text{rotation}} \\leq \\min(T_{\\text{policy}}, T_{\\text{compromise\\_risk}})`}
  variables={[
    { symbol: "T_{\\text{rotation}}", description: "Time between secret rotations" },
    { symbol: "T_{\\text{policy}}", description: "Compliance policy requirement" },
    { symbol: "T_{\\text{compromise\_risk}}", description: "Risk-based rotation interval" }
  ]}
/>

<MathNote type="info">
Never store secrets in DAG files or environment variables in plaintext. Use Airflow's Secrets Backend (Vault, AWS Secrets Manager, GCP Secret Manager) for all credentials.
</MathNote>

<MathNote type="tip">
Enable audit logging to track who accessed what secrets and when. This is required for SOC2 and GDPR compliance.
</MathNote>

## Key Concepts Table

| Security Layer | Component | Implementation | Priority |
|----------------|-----------|----------------|----------|
| **Authentication** | Webserver | OAuth, LDAP, SAML | P0 |
| **Authorization** | RBAC | Role-based permissions | P0 |
| **Encryption** | Database | TLS, AES-256 | P0 |
| **Secrets** | Backend | Vault, AWS SM | P0 |
| **Network** | Infrastructure | Firewalls, VPN | P1 |
| **Audit** | Logging | Activity logs | P1 |
| **Compliance** | Policies | SOC2, GDPR | P2 |

## Code Examples

### Security Audit Script

```python
# security_audit.py
from airflow import settings
from airflow.models import Connection, Variable
from sqlalchemy import text
import json

def audit_security_posture():
    """Comprehensive security audit of Airflow deployment."""
    session = settings.Session()
    findings = []
    
    # Check for plaintext passwords in metadata DB
    plaintext_conns = session.query(Connection).filter(
        Connection.conn_type.notin_(['aws', 'google_cloud_platform'])
    ).all()
    
    for conn in plaintext_conns:
        if conn.password and not conn.password.startswith('&#123;'):
            findings.append(&#123;
                'severity': 'HIGH',
                'category': 'Secrets',
                'finding': f'Plaintext password in connection: &#123;conn.conn_id&#125;',
                'recommendation': 'Move to secrets backend',
            &#125;)
    
    # Check for variables with sensitive data
    sensitive_patterns = ['password', 'secret', 'key', 'token']
    variables = session.query(Variable).all()
    
    for var in variables:
        if any(pattern in var.key.lower() for pattern in sensitive_patterns):
            findings.append(&#123;
                'severity': 'MEDIUM',
                'category': 'Secrets',
                'finding': f'Potentially sensitive variable: &#123;var.key&#125;',
                'recommendation': 'Move to secrets backend',
            &#125;)
    
    # Check for DAGs with hardcoded secrets
    import ast
    import os
    
    dags_folder = '/opt/airflow/dags'
    for root, dirs, files in os.walk(dags_folder):
        for file in files:
            if file.endswith('.py'):
                filepath = os.path.join(root, file)
                with open(filepath, 'r') as f:
                    content = f.read()
                
                # Simple pattern matching
                for pattern in ['password=', 'secret=', 'api_key=']:
                    if pattern in content:
                        findings.append(&#123;
                            'severity': 'HIGH',
                            'category': 'Code',
                            'finding': f'Potential hardcoded secret in &#123;filepath&#125;',
                            'recommendation': 'Use variables or secrets backend',
                        &#125;)
    
    return findings

def generate_security_report(findings):
    """Generate security audit report."""
    report = &#123;
        'total_findings': len(findings),
        'high_severity': len([f for f in findings if f['severity'] == 'HIGH']),
        'medium_severity': len([f for f in findings if f['severity'] == 'MEDIUM']),
        'low_severity': len([f for f in findings if f['severity'] == 'LOW']),
        'findings': findings,
    &#125;
    
    print(json.dumps(report, indent=2))
    return report

if __name__ == "__main__":
    findings = audit_security_posture()
    generate_security_report(findings)