fix: restore proxy syntax and adversarial signature checks

.github/PULL_REQUEST_TEMPLATE.md

@@ -0,0 +1,15 @@
+## Problem and change
+
+Describe the concrete problem and resulting behavior. Link an issue if applicable.
+
+## Validation
+
+List the commands or manual checks run and their results. Explain checks not run.
+
+## Review checklist
+
+- [ ] The branch targets the current default branch and preserves unrelated work.
+- [ ] Documentation and configuration match the implemented behavior.
+- [ ] Tests cover changed behavior where appropriate.
+- [ ] No credentials, personal records, or client data are included.
+- [ ] Deployment, migration, dependency, and rollback implications are documented where relevant.

.github/workflows/quality.yml

@@ -62,3 +62,18 @@ jobs:
           languages: python
       - uses: github/codeql-action/autobuild@v3
       - uses: github/codeql-action/analyze@v3
+
+  terraform:
+    name: Terraform validation
+    runs-on: ubuntu-24.04
+    steps:
+      - uses: actions/checkout@v4
+      - uses: hashicorp/setup-terraform@v3
+      - name: Format
+        run: terraform fmt -check -recursive terraform
+      - name: Initialise without backend
+        working-directory: terraform
+        run: terraform init -backend=false -input=false
+      - name: Validate
+        working-directory: terraform
+        run: terraform validate

proxy/config.py

@@ -2,8 +2,6 @@
 Configuration for the LLM Security Proxy.
 """
 
-from typing import Optional
-
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -22,13 +20,13 @@ class Settings(BaseSettings):
 
     # Client authentication. Required outside development unless explicitly disabled.
     REQUIRE_API_KEY: bool = True
-    PROXY_API_KEY: Optional[str] = None
+    PROXY_API_KEY: str | None = None
 
     # Upstream providers
     OPENAI_API_BASE: str = "https://api.openai.com/v1"
-    OPENAI_API_KEY: Optional[str] = None
+    OPENAI_API_KEY: str | None = None
     ANTHROPIC_API_BASE: str = "https://api.anthropic.com/v1"
-    ANTHROPIC_API_KEY: Optional[str] = None
+    ANTHROPIC_API_KEY: str | None = None
 
     # Request limits / budgets
     MAX_PROMPT_TOKENS: int = 4096
@@ -52,7 +50,7 @@ class Settings(BaseSettings):
     # Audit
     AUDIT_LOG_PATH: str = "/var/log/llm-security-proxy/audit.log"
     SIGN_AUDIT_LOGS: bool = False
-    LOG_SIGNING_KEY: Optional[str] = None
+    LOG_SIGNING_KEY: str | None = None
 
 
 settings = Settings()

proxy/guardrails.py

@@ -3,28 +3,27 @@
 """
 
 import re
-from typing import List, Tuple
 
 from proxy.config import settings
 
 
 class GuardrailEngine:
     def __init__(self):
-        self.injection_keywords: List[str] = [
-            r"ignores+(?:alls+)?previouss+instructions",
-            r"systems+(?:override|bypass)",
-            r"yous+ares+nows+(?:a|an)s+unrestricted",
-            r"developers+mode",
-            r"dos+anythings+nows+dan",
-            r"acts+ass+as+computers+terminals+withs+nos+restrictions",
+        self.injection_keywords: list[str] = [
+            r"ignore\s+(?:all\s+)?previous\s+instructions",
+            r"system\s+(?:override|bypass)",
+            r"you\s+are\s+now\s+(?:a|an)\s+unrestricted",
+            r"developer\s+mode",
+            r"do\s+anything\s+now\s+dan",
+            r"act\s+as\s+a\s+computer\s+terminal\s+with\s+no\s+restrictions",
             r"jailbreak",
-            r"bypasss+safetys+filters",
+            r"bypass\s+safety\s+filters",
         ]
         self.compiled_keywords = [
             re.compile(pattern, re.IGNORECASE) for pattern in self.injection_keywords
         ]
 
-    def check_injection(self, prompt: str) -> Tuple[bool, float, str]:
+    def check_injection(self, prompt: str) -> tuple[bool, float, str]:
         if not settings.ENABLE_GUARDRAILS:
             return False, 0.0, "Guardrails are disabled"
 
@@ -49,5 +48,5 @@ def check_injection(self, prompt: str) -> Tuple[bool, float, str]:
 
         return False, 0.0, "Prompt passed security filters"
 
-    def _run_ml_guard_inference(self, prompt: str) -> Tuple[bool, float, str]:
+    def _run_ml_guard_inference(self, prompt: str) -> tuple[bool, float, str]:
         return False, 0.0, ""

proxy/logger.py

@@ -14,12 +14,13 @@
 copies or substantial portions of the Software.
 """
 
-import hmac
 import hashlib
+import hmac
 import json
 import logging
 import time
-from typing import Dict, Any, Optional
+from typing import Any
+
 from proxy.config import settings
 
 
@@ -32,32 +33,32 @@ class AuditLogger:
     def __init__(self):
         self.logger = logging.getLogger("llm-security-proxy-audit")
         self.logger.setLevel(logging.INFO)
-        
+
         # Configure file handler for compliance audit trail
         try:
             handler = logging.FileHandler(settings.AUDIT_LOG_PATH)
-            formatter = logging.Formatter('%(message)s')
+            formatter = logging.Formatter("%(message)s")
             handler.setFormatter(formatter)
             self.logger.addHandler(handler)
-        except Exception:
+        except OSError:
             # Fallback to stdout logging in case directory is missing or permissions fail
             handler = logging.StreamHandler()
-            formatter = logging.Formatter('%(message)s')
+            formatter = logging.Formatter("%(message)s")
             handler.setFormatter(formatter)
             self.logger.addHandler(handler)
 
     def log_security_event(
         self,
         event_type: str,
-        user_id: Optional[str],
-        request_metadata: Dict[str, Any],
+        user_id: str | None,
+        request_metadata: dict[str, Any],
         redacted_entities: list,
         guardrail_score: float,
-        status: str
+        status: str,
     ) -> None:
         """
         Creates, signs, and writes a compliance-level security audit event.
-        
+
         Args:
             event_type: The class of transaction (e.g., chat.completion, system.configuration).
             user_id: The ID of the enterprise user triggering the event.
@@ -77,25 +78,25 @@ def log_security_event(
             "guardrail_score": guardrail_score,
             "status": status,
             "app_name": settings.APP_NAME,
-            "app_env": settings.APP_ENV
+            "app_env": settings.APP_ENV,
         }
 
         # Cryptographic sign log payload if key is configured (SOC2 / HIPAA compliance)
         if settings.SIGN_AUDIT_LOGS and settings.LOG_SIGNING_KEY:
-            signature = self._generate_hmac_signature(event_payload, settings.LOG_SIGNING_KEY)
+            signature = self._generate_hmac_signature(
+                event_payload, settings.LOG_SIGNING_KEY
+            )
             event_payload["cryptographic_signature"] = signature
             event_payload["signature_algorithm"] = "HMAC-SHA256"
 
         # Log as single line JSON string for SIEM parser ingestion
         self.logger.info(json.dumps(event_payload))
 
-    def _generate_hmac_signature(self, payload: Dict[str, Any], secret_key: str) -> str:
+    def _generate_hmac_signature(self, payload: dict[str, Any], secret_key: str) -> str:
         """
         Generates HMAC-SHA256 signature to guarantee log integrity.
         """
-        serialized = json.dumps(payload, sort_keys=True).encode('utf-8')
+        serialized = json.dumps(payload, sort_keys=True).encode("utf-8")
         return hmac.new(
-            secret_key.encode('utf-8'),
-            serialized,
-            hashlib.sha256
+            secret_key.encode("utf-8"), serialized, hashlib.sha256
         ).hexdigest()

proxy/main.py

@@ -4,7 +4,7 @@
 
 import time
 import uuid
-from typing import Any, Dict, List, Optional
+from typing import Any
 
 import httpx
 from fastapi import FastAPI, Header, HTTPException, Request, status
@@ -43,8 +43,10 @@ class ChatMessage(BaseModel):
 
 class ChatCompletionRequest(BaseModel):
     model: str = Field(..., min_length=1, max_length=200)
-    messages: List[ChatMessage] = Field(..., min_length=1, max_length=settings.MAX_MESSAGES)
-    temperature: Optional[float] = Field(default=1.0, ge=0.0, le=2.0)
+    messages: list[ChatMessage] = Field(
+        ..., min_length=1, max_length=settings.MAX_MESSAGES
+    )
+    temperature: float | None = Field(default=1.0, ge=0.0, le=2.0)
     stream: bool = False
 
 
@@ -74,9 +76,8 @@ def _authenticate(x_api_key: str | None, authorization: str | None) -> None:
         )
 
 
-def _prompt_text(messages: List[ChatMessage]) -> str:
-    return "
-".join(msg.content for msg in messages)
+def _prompt_text(messages: list[ChatMessage]) -> str:
+    return "\n".join(msg.content for msg in messages)
 
 
 @app.middleware("http")
@@ -122,9 +123,7 @@ async def health_check():
 
 @app.get("/readyz", status_code=status.HTTP_200_OK)
 async def readiness_check():
-    upstream_configured = bool(
-        settings.OPENAI_API_KEY or settings.ANTHROPIC_API_KEY
-    )
+    upstream_configured = bool(settings.OPENAI_API_KEY or settings.ANTHROPIC_API_KEY)
     return {
         "status": "ready",
         "upstream_configured": upstream_configured,
@@ -147,10 +146,13 @@ async def proxy_chat_completions(
     if len(full_prompt_text) > settings.MAX_MESSAGE_CHARS:
         raise HTTPException(
             status_code=413,
-            detail={"error": "Prompt exceeds configured character limit.", "request_id": request_id},
+            detail={
+                "error": "Prompt exceeds configured character limit.",
+                "request_id": request_id,
+            },
         )
 
-    metadata: Dict[str, Any] = {
+    metadata: dict[str, Any] = {
         "ip_address": request.client.host if request.client else "unknown",
         "user_agent": request.headers.get("user-agent", "unknown"),
         "request_id": request_id,
@@ -177,7 +179,7 @@ async def proxy_chat_completions(
             },
         )
 
-    redacted_messages: List[ChatMessage] = []
+    redacted_messages: list[ChatMessage] = []
     total_redacted_info: list = []
 
     for message in payload.messages: