This document describes the security model of the AI Customer Support Agent, including current implementations, known demo shortcuts, and recommendations for production hardening.

The system uses stateless JWT authentication via the python-jose library.

How it works:

1. User submits credentials to POST /auth/login
2. Backend verifies email/password against the users table
3. On success, a JWT is generated with payload:

   {
     "sub": "USER-001",
     "role": "customer",
     "exp": 1705312200
   }

4. Token is signed with HS256 algorithm and the configured JWT_SECRET
5. Token expires after 24 hours
6. Client includes the token in the Authorization: Bearer <token> header on all protected requests

Implementation file: mini-services/refund-backend/routes/auth.py

The get_current_user FastAPI dependency:

1. Extracts the Authorization header from the incoming Request
2. Validates the Bearer prefix
3. Decodes the JWT using the configured secret and algorithm
4. Checks expiration (handled by python-jose)
5. Looks up the user in the database (ensures user still exists)
6. Returns the User ORM object for use in route handlers

Two roles are defined: admin and customer.

• get_current_user: FastAPI dependency that extracts and validates the JWT, returns User object
• require_admin: FastAPI dependency that wraps get_current_user and checks user.role == "admin", returning 403 if not
• Route-level filtering: Customer routes filter queries by current_user.customer_id to enforce data isolation

# Example: Customer can only see their own refunds
if current_user.role == "customer":
    q = q.filter(RefundRequest.customer_id == current_user.customer_id)

All API request bodies are validated by Pydantic models in schemas.py:

• LoginRequest: Requires email (string) and password (string)
• ChatRequest: Requires message (string) and user_id (string); optional conversation_id
• RefundRequestCreate: Requires order_id, customer_id, reason; optional amount
• RefundPolicyUpdate: All fields optional (value, is_active, description)

Pydantic automatically rejects:

• Missing required fields (400 Bad Request)
• Wrong data types (422 Unprocessable Entity)
• Extra fields not in the schema

FastAPI's Query type is used for bounded parameters:

@router.get("", response_model=list[AuditLogOut])
def list_logs(
    limit: int = Query(100, ge=1, le=1000),  # Must be 1-1000
    offset: int = Query(0, ge=0),            # Must be non-negative
    ...

Path parameters (e.g., {refund_id}, {customer_id}) are typed as strings. The backend queries the database and returns 404 if not found — there is no risk of path traversal since SQLAlchemy uses parameterized queries.

Status: ✅ Fully protected

The entire application uses SQLAlchemy ORM with parameterized queries. There is no raw SQL anywhere in the codebase.

# Safe — SQLAlchemy parameterizes this
user = db.query(User).filter(User.email == body.email).first()

# The generated SQL is:
# SELECT * FROM users WHERE email = ?  -- with body.email as a bound parameter

The filter() method never interpolates user input into the SQL string. Even if a user submitted a malicious email like ' OR 1=1 --, SQLAlchemy would treat it as a literal string value.

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],      # All origins allowed
    allow_credentials=True,
    allow_methods=["*"],       # All HTTP methods
    allow_headers=["*"],       # All headers
)

Restrict allow_origins to the actual frontend domain(s):

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "https://shopmart.example.com",
        "https://admin.shopmart.example.com",
    ],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT"],
    allow_headers=["Authorization", "Content-Type"],
)

Alternatively, use an environment variable:

import os
ALLOWED_ORIGINS = os.getenv("CORS_ORIGINS", "http://localhost:3000").split(",")
app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT"],
    allow_headers=["Authorization", "Content-Type"],
)

Passwords are stored and compared as plaintext in the users table.

# In routes/auth.py — DEMO ONLY
if not user or user.password != body.password:
    raise HTTPException(status_code=401)

This is intentional for demo convenience. All demo passwords are simple (admin123, password1, etc.) so presenters can type them quickly.

Use bcrypt for password hashing:

pip install bcrypt

from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# On user creation:
hashed_password = pwd_context.hash(plain_password)

# On login verification:
if not pwd_context.verify(body.password, user.password):
    raise HTTPException(status_code=401)

Update the seed data to use hashed passwords:

# In seed.py
hashed = pwd_context.hash("admin123")
db.add(User(id="USER-ADMIN", email="admin@shopmart.com", password=hashed, ...))

• Secret: Hardcoded as "demo-secret-key-change-in-production" in config.py
• Algorithm: HS256
• Expiry: 24 hours
• No refresh tokens

1. Use a strong, random secret (minimum 32 characters):

   import secrets
   # Generate: secrets.token_urlsafe(32)
   JWT_SECRET = os.getenv("JWT_SECRET")  # Must be set

2. Add refresh tokens for longer sessions:

   • Issue a short-lived access token (15 min) and a long-lived refresh token (7 days)
   • Implement POST /auth/refresh to exchange a refresh token for a new access token
   • Store refresh tokens in the database for revocation capability

3. Consider rotating secrets periodically

4. Validate the alg header to prevent algorithm confusion attacks (python-jose does this by default)

There is no rate limiting on any endpoint.

Install slowapi:

pip install slowapi

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter

# Apply to login endpoint
@app.post("/auth/login")
@limiter.limit("5/minute")  # 5 login attempts per minute per IP
def login(request: Request, body: LoginRequest, ...):
    ...

# Apply to chat endpoint
@app.post("/chat")
@limiter.limit("10/minute")  # 10 messages per minute per user
def chat(request: Request, body: ChatRequest, ...):
    ...

Key endpoints to rate-limit:

• POST /auth/login — Prevent brute-force attacks (5-10/minute)
• POST /chat — Prevent abuse and LLM resource exhaustion (10-20/minute)

1. Prompt injection: A user could attempt to manipulate the system prompt
2. Tool misuse: The LLM might call tools with unexpected parameters

1. System prompt is server-side — Users cannot modify it
2. Tool parameters are validated by Pydantic schemas and the tool functions themselves
3. Tools return structured data — The agent makes decisions based on tool outputs, not user claims
4. No destructive tools — Tools only read data and create records; nothing is deleted or modified except new inserts
5. Policy checks are code-enforced — Even if the LLM approves something, the tool layer enforces constraints (e.g., check_refund_window queries the database, doesn't trust the LLM)

1. Sanitize user input before including in LLM context (strip excessively long messages)
2. Add a maximum message length to prevent token-bombing
3. Log all LLM inputs/outputs for security auditing (already done via AgentExecution)
4. Consider output filtering to prevent the LLM from leaking system information