docs/security.md

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:

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:

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:

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:

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:

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:

# 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:

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:

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:

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	✅