Stream D: Security hardening — 10 fixes from QA review

- SEC-05: Fail fast if SECRET_KEY is default 'change-me-in-production'
- SEC-06: CORS allow_credentials=False (Bearer tokens don't need it)
- SEC-08: Input validation on auth schemas (min/max length, username pattern)
- SEC-10: .dockerignore files (root, backend, mcp-connector)
- SEC-11: Hardcode ALGORITHM='HS256' as constant, remove from settings
- CQ-01: Gate crawl scheduler on CRAWL_ENABLED env var (default false)
- OS-02: CONTRIBUTING.md
- OS-03: CHANGELOG.md with v1.0.0 entry
- setup.ps1: Add API_TOKEN generation (Windows parity with setup.sh)
This commit is contained in:
Claude 2026-06-21 23:13:53 +00:00
parent c075eb9257
commit 6e082ce58f
10 changed files with 113 additions and 20 deletions

11
backend/.dockerignore Normal file
View file

@ -0,0 +1,11 @@
.git
.env
.env.*
!.env.example
__pycache__
*.pyc
content/
node_modules/
dist/
*.log
.DS_Store

View file

@ -10,7 +10,7 @@ class Settings(BaseSettings):
# Auth
secret_key: str = "change-me-in-production"
algorithm: str = "HS256"
crawl_enabled: bool = False # Set CRAWL_ENABLED=true to start program crawl scheduler
access_token_expire_minutes: int = 1440 # 24 hours
api_token: str = "" # MCP connector auth — generated by setup.sh

View file

@ -1,6 +1,7 @@
from __future__ import annotations
import asyncio
import sys
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI
@ -27,6 +28,13 @@ async def lifespan(app: FastAPI):
logger.error("Could not initialize database after 10 attempts — starting without tables")
# SEC-05: Fail fast if SECRET_KEY not configured
from app.config import get_settings
_s = get_settings()
if _s.secret_key == "change-me-in-production":
logger.error("CRITICAL: SECRET_KEY not set. Run ./setup.sh or set SECRET_KEY in .env")
sys.exit(1)
# Run lightweight migrations for schema changes
try:
await run_migrations()
@ -41,14 +49,17 @@ async def lifespan(app: FastAPI):
except Exception as e:
logger.warning(f"Resource seeding failed (non-fatal): {e}")
# Start background crawl queue scheduler
# Start background crawl queue scheduler (gated on CRAWL_ENABLED)
crawl_task = None
try:
from app.services.crawl_scheduler import crawl_queue_loop
crawl_task = asyncio.create_task(crawl_queue_loop())
logger.info("Crawl queue scheduler started")
except Exception as e:
logger.warning(f"Crawl scheduler failed to start (non-fatal): {e}")
if _s.crawl_enabled:
try:
from app.services.crawl_scheduler import crawl_queue_loop
crawl_task = asyncio.create_task(crawl_queue_loop())
logger.info("Crawl queue scheduler started")
except Exception as e:
logger.warning(f"Crawl scheduler failed to start (non-fatal): {e}")
else:
logger.info("Crawl scheduler disabled (set CRAWL_ENABLED=true to enable)")
logger.info("Fitness Rewards backend started")
yield
@ -68,7 +79,7 @@ app = FastAPI(title="Fitness Rewards", version="1.0.0", lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_credentials=False # Bearer tokens don't need credentials mode,
allow_methods=["*"],
allow_headers=["*"],
)

View file

@ -1,17 +1,17 @@
from __future__ import annotations
from pydantic import BaseModel
from pydantic import BaseModel, Field
class LoginRequest(BaseModel):
username: str
password: str
username: str = Field(min_length=3, max_length=50)
password: str = Field(min_length=8, max_length=128)
class RegisterRequest(BaseModel):
username: str
password: str
display_name: str
username: str = Field(min_length=3, max_length=50, pattern=r'^[a-zA-Z0-9_-]+$')
password: str = Field(min_length=8, max_length=128)
display_name: str = Field(min_length=1, max_length=100)
email: str | None = None

View file

@ -13,6 +13,7 @@ from app.config import get_settings
from app.database import get_db
settings = get_settings()
ALGORITHM = "HS256" # Hardcoded — not configurable for security
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login", auto_error=False)
@ -27,7 +28,7 @@ def verify_password(plain: str, hashed: str) -> bool:
def create_access_token(user_id: str, expires_delta: timedelta | None = None) -> str:
expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=settings.access_token_expire_minutes))
payload = {"sub": user_id, "exp": expire}
return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm)
return jwt.encode(payload, settings.secret_key, algorithm=ALGORITHM)
async def get_current_user(