# Security

Security architecture for production deployment.

## Overview

FixMyNeighborhood uses a **hybrid security model**:

| Layer | Responsibility | Why |
|-------|---------------|-----|
| **Python** | Rate limiting, input validation, prompt injection | Deterministic, fast, can't be jailbroken |
| **LLM** | Business validation, user interaction | Flexible, contextual, intelligent |

```
User Input
    │
    ▼
┌─────────────────────────────────────┐
│         PYTHON SECURITY LAYER       │
│                                     │
│  1. Rate Limiting (abuse prevention)│
│  2. Input Validation (sanitization) │
│  3. Prompt Injection (detection)    │
│                                     │
│  Blocks: Abuse, garbage, attacks    │
│  Passes: Clean input to LLM         │
└─────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────┐
│         LLM BUSINESS LAYER          │
│                                     │
│  1. Is this infrastructure?         │
│  2. Is location valid?              │
│  3. What priority?                  │
│  4. What follow-up questions?       │
│                                     │
│  Full autonomy for decisions        │
└─────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────┐
│         OUTPUT SECURITY LAYER       │
│                                     │
│  1. PII Masking (logs)              │
│  2. Audit Trail (compliance)        │
│                                     │
└─────────────────────────────────────┘
```

## Rate Limiting

**File**: `security/rate_limiter.py`

Sliding window rate limiting per session:

```python
class RateLimiter:
    def __init__(
        self,
        max_requests: int = 10,
        window_seconds: int = 60,
        min_interval_ms: int = 2000,
    ):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.min_interval_ms = min_interval_ms
```

**Configuration**:

| Parameter | Default | Description |
|-----------|---------|-------------|
| `max_requests` | 10 | Max requests per window |
| `window_seconds` | 60 | Sliding window duration |
| `min_interval_ms` | 2000 | Minimum time between requests |

**Usage in app.py**:

```python
try:
    rate_limiter.check_rate_limit(session_id)
except RateLimitExceeded as e:
    # Return friendly message, don't process
    return f"Please wait {e.wait_seconds} seconds"
```

## Input Validation

**File**: `security/input_validator.py`

Validates and sanitizes user input:

```python
class InputValidator:
    def validate_text(self, text: str, context: str = "message") -> ValidationResult:
        # Check length
        if len(text) > self.max_length:
            return ValidationResult(is_valid=False, error="too_long")

        # Check for HTML/XSS
        if self._contains_html(text):
            sanitized = self._strip_html(text)
            return ValidationResult(
                is_valid=True,
                sanitized_value=sanitized,
                warning="html_stripped"
            )
```

**Checks**:

| Check | Action |
|-------|--------|
| Empty input | Block (unless image provided) |
| Too long (>5000 chars) | Block |
| HTML/script tags | Strip and warn |
| Excessive whitespace | Normalize |

## Prompt Injection Protection

**File**: `security/prompt_guard.py`

Detects and blocks prompt injection attempts:

```python
class PromptGuard:
    PATTERNS = {
        "role_impersonation": [
            r"you are now",
            r"ignore (?:all )?(?:previous|above)",
            r"disregard (?:all )?(?:previous|above)",
            r"forget (?:all )?(?:previous|above)",
        ],
        "instruction_override": [
            r"your (?:new )?instructions are",
            r"system prompt:",
            r"admin override",
        ],
        "jailbreak": [
            r"DAN mode",
            r"developer mode",
            r"pretend you",
            r"act as if",
        ],
    }
```

**Threat Levels**:

| Level | Action | Example |
|-------|--------|---------|
| `none` | Allow | "pothole on Broadway" |
| `low` | Allow, log | Contains "ignore" but in context |
| `medium` | Allow, sanitize | Suspicious but not malicious |
| `high` | Block | Clear jailbreak attempt |

**Response for blocked input**:

```python
if not guard_result.is_safe:
    return "Please describe your infrastructure issue normally."
```

## Geographic Validation

Geographic validation (NYC bounds checking) is handled by the **MCP tools** rather than a separate Python security layer:

- `validate_address` - Validates addresses via NYC GeoSearch API
- `geo_search_address` - Reverse geocoding via Photon API

This approach lets the LLM make intelligent decisions about location validation, using real NYC API data rather than static bounding box checks.

## Cross-User Isolation

**Mechanism**: Gradio `gr.State`

Each user session has isolated state:

```python
# In app.py
session_logs = gr.State([])           # Per-session logs
session_orchestrator = gr.State(None)  # Per-session orchestrator
```

**Why this works**:
- `gr.State` is tied to browser session
- No shared mutable state between users
- Orchestrator maintains per-session conversation history

**What's isolated**:

| Component | Isolation |
|-----------|-----------|
| Conversation history | Per-session |
| Rate limit counters | Per-session |
| Uploaded images | Unique filenames |
| Logs | Per-session capture |

## Output Masking

**File**: `security/output_masker.py`

Masks PII in logs and audit trails:

```python
class OutputMasker:
    PATTERNS = {
        "email": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
        "phone": r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b",
        "ssn": r"\b\d{3}-\d{2}-\d{4}\b",
    }

    def mask(self, text: str) -> str:
        for pattern_name, pattern in self.PATTERNS.items():
            text = re.sub(pattern, f"[{pattern_name.upper()}_MASKED]", text)
        return text
```

**Masked in logs**:
- Email addresses → `[EMAIL_MASKED]`
- Phone numbers → `[PHONE_MASKED]`
- SSN patterns → `[SSN_MASKED]`

## Error Handling

**File**: `security/error_handler.py`

Converts technical errors to user-friendly messages:

```python
class ErrorHandler:
    ERROR_MESSAGES = {
        "rate_limit": "You're sending requests too quickly. Please wait a moment.",
        "api_error": "We're having trouble connecting to our services. Please try again.",
        "validation": "Please check your input and try again.",
    }

    def handle(self, error: Exception, context: str = "") -> FriendlyError:
        # Map technical error to friendly message
        # Log full error for debugging
        # Return safe message for user
```

**Principles**:
- Never expose stack traces to users
- Log full errors for debugging
- Provide actionable user messages

## HTTPS

All external calls use HTTPS:

| Service | URL | Protocol |
|---------|-----|----------|
| MCP Server | `https://...hf.space` | HTTPS |
| Weather.gov | `https://api.weather.gov` | HTTPS |
| NYC GeoSearch | `https://geosearch.planninglabs.nyc` | HTTPS |
| Photon API | `https://photon.komoot.io` | HTTPS |
| NYC Open Data | `https://data.cityofnewyork.us` | HTTPS |
| Resend API | `https://api.resend.com` | HTTPS |

## Audit Trail

**File**: `observability/audit_trail.py`

Logs all requests for compliance:

```python
class AuditTrail:
    def start_request(self, session_id: str, input_preview: str, has_image: bool) -> str:
        """Start tracking a request."""
        request_id = str(uuid.uuid4())[:8]
        self._log({
            "event": "request_start",
            "request_id": request_id,
            "session_id": session_id[:8],  # Truncated for privacy
            "has_image": has_image,
            "timestamp": time.time(),
        })
        return request_id
```

**Logged events**:
- Request start/end
- Rate limit hits
- Validation failures
- Security blocks
- Errors

## Security Checklist

| Requirement | Implementation | Status |
|-------------|----------------|--------|
| Rate limiting | `security/rate_limiter.py` | ✅ |
| Input validation | `security/input_validator.py` | ✅ |
| Prompt injection | `security/prompt_guard.py` | ✅ |
| Cross-user isolation | `gr.State` per-session | ✅ |
| HTTPS for APIs | All external calls | ✅ |
| PII masking | `security/output_masker.py` | ✅ |
| Error handling | `security/error_handler.py` | ✅ |
| Audit logging | `observability/audit_trail.py` | ✅ |