feat: address backend security audit findings

.env.example

@@ -5,6 +5,16 @@ ENVIRONMENT="local"
 FIRST_SUPERUSER="admin@example.com"
 FIRST_SUPERUSER_PASSWORD="changethis"
 FRONTEND_HOST="http://localhost:5173"
+# Comma-separated Host header allowlist enforced by TrustedHostMiddleware in
+# production (e.g. "api.example.com,example.com"). Ignored outside production,
+# where all hosts are allowed for local dev. The frontend host is added
+# automatically.
+ALLOWED_HOSTS=""
+
+# Host address Docker binds the DB/Redis ports to. Defaults to loopback
+# (127.0.0.1) so they are not exposed on the host's network interfaces. Set to
+# 0.0.0.0 or a specific IP only if you need external access.
+DOCKER_BIND_HOST="127.0.0.1"
 
 # Database Settings
 POSTGRES_SERVER="localhost"

README.md

@@ -94,6 +94,11 @@ ENVIRONMENT="local"
 FIRST_SUPERUSER="admin@example.com"
 FIRST_SUPERUSER_PASSWORD="changethis"
 FRONTEND_HOST="http://localhost:5173"
+# Comma-separated Host header allowlist enforced in production only.
+ALLOWED_HOSTS=""
+
+# Host address Docker binds the DB/Redis ports to (loopback by default).
+DOCKER_BIND_HOST="127.0.0.1"
 
 # Database Settings
 POSTGRES_SERVER="localhost"

app/api/routes/auth.py

@@ -4,7 +4,7 @@
 from fastapi.security import OAuth2PasswordRequestForm
 
 from app.api.decorators import audit_unexpected_failure
-from app.api.deps import CurrentActiveUser, SessionDep
+from app.api.deps import CurrentActiveUser, SessionDep, reusable_oauth2
 from app.core.config import settings
 from app.core.messages.error_message import ErrorMessages
 from app.core.messages.success_message import SuccessMessages
@@ -18,7 +18,7 @@
     ForgotPassword,
     NewPassword,
     UpdatePassword,
-    UserCreate,
+    UserRegister,
     VerifyEmail,
 )
 from app.schemas.user_activity import ActivityType, ResourceType
@@ -36,10 +36,11 @@
 
 router = APIRouter()
 
-# The refresh cookie is limited to the refresh endpoint only so it is never
-# sent on unrelated requests. This path must match for both set_cookie and
-# delete_cookie calls or the cookie cannot be cleared.
-_REFRESH_COOKIE_PATH = f"{settings.API_V1_STR}/auth/refresh"
+# The refresh cookie is scoped to the auth router so it is sent on the auth
+# endpoints that must revoke it — /refresh (rotation), /logout, and
+# /change-password — but not on the rest of the API. This path must match for
+# both set_cookie and delete_cookie calls or the cookie cannot be cleared.
+_REFRESH_COOKIE_PATH = f"{settings.API_V1_STR}/auth"
 _COOKIE_SECURE = settings.ENVIRONMENT != "local"
 
 
@@ -128,6 +129,7 @@ async def refresh_token(
         request=request, session=session, refresh_token=refresh_token_cookie
     )
     _set_access_cookie(response, result.access_token)
+    _set_refresh_cookie(response, result.refresh_token)
     return CookieRefreshResponse(message=result.message)
 
 
@@ -138,11 +140,15 @@ async def refresh_token(
     endpoint="/logout",
 )
 async def logout(request: Request, response: Response, session: SessionDep) -> Message:
-    """Clear refresh token cookie and invalidate token in the blacklist."""
+    """Clear the auth cookies and revoke both tokens in the blacklist."""
     refresh_token_cookie = request.cookies.get("refresh_token")
-    if refresh_token_cookie:
+    access_token_cookie = request.cookies.get("access_token")
+    if refresh_token_cookie or access_token_cookie:
         await logout_service(
-            request=request, session=session, refresh_token=refresh_token_cookie
+            request=request,
+            session=session,
+            refresh_token=refresh_token_cookie,
+            access_token=access_token_cookie,
         )
     _clear_auth_cookies(response)
     return Message(success=True, message=SuccessMessages.LOGOUT_SUCCESS)
@@ -156,10 +162,10 @@ async def logout(request: Request, response: Response, session: SessionDep) -> M
     endpoint="/register",
 )
 async def register_user(
-    request: Request, session: SessionDep, user_in: UserCreate
+    request: Request, session: SessionDep, user_in: UserRegister
 ) -> Message:
     """Register a new user."""
-    await register_service(request=request, session=session, user_create=user_in)
+    await register_service(request=request, session=session, user_register=user_in)
     return Message(success=True, message=SuccessMessages.REGISTER_SUCCESS)
 
 
@@ -243,14 +249,23 @@ async def resend_verification(
 )
 async def change_password(
     request: Request,
+    response: Response,
     session: SessionDep,
     current_user: CurrentActiveUser,
     body: UpdatePassword,
+    bearer_token: Annotated[str | None, Depends(reusable_oauth2)] = None,
 ) -> Message:
-    """Change user password while logged in."""
-    return await change_password_service(
+    """Change user password while logged in, then end the current session."""
+    # Resolve the access token the same way ``get_current_user`` does so the
+    # session is revoked for Bearer clients too, not only cookie-based ones.
+    access_token = request.cookies.get("access_token") or bearer_token
+    result = await change_password_service(
         request=request,
         session=session,
         current_user=current_user,
         update_password=body,
+        access_token=access_token,
+        refresh_token=request.cookies.get("refresh_token"),
     )
+    _clear_auth_cookies(response)
+    return result

app/core/config.py

@@ -29,11 +29,18 @@ class Settings(BaseSettings):
     )
     API_V1_STR: str = "/api/v1"
     SECRET_KEY: str = secrets.token_urlsafe(32)
-    # 8 days — long enough to avoid surprise logouts, short enough to limit
-    # blast radius of a stolen token when combined with refresh rotation.
-    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8
+    # 60 minutes — short-lived access token limits the blast radius of a
+    # stolen token. Clients transparently renew via the refresh-token flow.
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60
     REFRESH_TOKEN_EXPIRE_DAYS: int = 30
     EMAIL_RESET_TOKEN_EXPIRE_HOURS: int = 48
+
+    # Account lockout: after MAX failed logins within the WINDOW, the account is
+    # locked for LOCKOUT seconds. Closes distributed brute-force that the
+    # per-IP rate limit cannot (attempts keyed by email, not source IP).
+    LOGIN_MAX_FAILED_ATTEMPTS: int = 5
+    LOGIN_FAILED_ATTEMPT_WINDOW_SECONDS: int = 15 * 60
+    LOGIN_LOCKOUT_SECONDS: int = 15 * 60
     FRONTEND_HOST: str = "http://localhost:5173"
     ENVIRONMENT: Literal["local", "staging", "production"] = "local"
     DEFAULT_LANGUAGE: Literal["en", "tr"] = "en"
@@ -50,6 +57,33 @@ def all_cors_origins(self) -> list[str]:
             origins.append(str(self.FRONTEND_HOST).rstrip("/"))
         return origins
 
+    # Host header allowlist enforced by TrustedHostMiddleware. Comma-separated
+    # in the environment (e.g. ``api.example.com,example.com``).
+    ALLOWED_HOSTS: Annotated[list[str] | str, BeforeValidator(parse_cors)] = []
+
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def trusted_hosts(self) -> list[str]:
+        """Hosts accepted by ``TrustedHostMiddleware``.
+
+        Outside production we return ``["*"]`` so local dev and the test client
+        (which sends ``Host: testserver``) are never blocked. Production
+        enforces the explicit ``ALLOWED_HOSTS`` list plus the frontend host,
+        falling back to ``["*"]`` only if nothing was configured so a missing
+        value can never brick the deployment.
+        """
+        if self.ENVIRONMENT != "production":
+            return ["*"]
+
+        from urllib.parse import urlparse
+
+        hosts = [h.strip() for h in self.ALLOWED_HOSTS if h.strip()]
+        if self.FRONTEND_HOST:
+            hostname = urlparse(str(self.FRONTEND_HOST)).hostname
+            if hostname:
+                hosts.append(hostname)
+        return hosts or ["*"]
+
     PROJECT_NAME: str
     SENTRY_DSN: str | None = None
 
@@ -120,10 +154,12 @@ def SQLALCHEMY_DATABASE_URI(self) -> PostgresDsn:
     FIRST_SUPERUSER_PASSWORD: str
 
     def _check_default_secret(self, var_name: str, value: str | None) -> None:
-        if value == "changethis":
+        is_blank = not (value or "").strip()
+        if value == "changethis" or is_blank:
+            reason = "empty" if is_blank else 'the insecure default "changethis"'
             message = (
-                f'The value of {var_name} is "changethis", '
-                "for security, please change it, at least for deployments."
+                f"The value of {var_name} is {reason}; "
+                "for security, please set it, at least for deployments."
             )
             if self.ENVIRONMENT == "local":
                 warnings.warn(message, stacklevel=1)

app/core/messages/error_message.py

@@ -14,6 +14,7 @@ class ErrorMessages:
     EMAIL_ALREADY_EXISTS = "error.user.email_exists"
     INVALID_CREDENTIALS = "error.user.invalid_credentials"
     INVALID_PASSWORD = "error.user.invalid_password"
+    WEAK_PASSWORD = "error.user.password_weak"
     USER_INACTIVE = "error.user.user_inactive"
     EMAIL_NOT_VERIFIED = "error.user.email_not_verified"
     INVALID_VERIFICATION_TOKEN = "error.user.invalid_verification_token"
@@ -32,6 +33,9 @@ class ErrorMessages:
     ACCOUNT_ALREADY_SUSPENDED = "error.account.already_suspended"
     ACCOUNT_NOT_SUSPENDED = "error.account.not_suspended"
 
+    # Account lockout (automatic, temporary — too many failed logins)
+    ACCOUNT_LOCKED = "error.account.locked"
+
     # Admin
     ADMIN_CANNOT_MODIFY_SELF = "error.admin.cannot_modify_self"
     ADMIN_CANNOT_DELETE_SELF = "error.admin.cannot_delete_self"

app/core/middleware.py

@@ -9,13 +9,33 @@
 from fastapi import FastAPI, Request
 from fastapi.responses import JSONResponse
 from starlette.middleware.cors import CORSMiddleware
+from starlette.middleware.trustedhost import TrustedHostMiddleware
 
 from app.core.config import settings
 from app.core.messages.error_message import ErrorMessages
 
 
 def register_middleware(app: FastAPI) -> None:
-    """Wire the origin check and (when configured) the CORS middleware."""
+    """Wire host validation, security headers, the origin check and CORS."""
+
+    @app.middleware("http")
+    async def security_headers_middleware(request: Request, call_next):
+        """Attach baseline security headers to every response.
+
+        Uses ``frame-ancestors 'none'`` rather than a full ``default-src``
+        policy so the bundled Swagger UI keeps loading its assets. HSTS is
+        only sent outside local since it requires HTTPS to be meaningful.
+        """
+        response = await call_next(request)
+        response.headers["X-Content-Type-Options"] = "nosniff"
+        response.headers["X-Frame-Options"] = "DENY"
+        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
+        response.headers["Content-Security-Policy"] = "frame-ancestors 'none'"
+        if settings.ENVIRONMENT != "local":
+            response.headers["Strict-Transport-Security"] = (
+                "max-age=63072000; includeSubDomains"
+            )
+        return response
 
     @app.middleware("http")
     async def origin_check_middleware(request: Request, call_next):
@@ -65,6 +85,17 @@ async def origin_check_middleware(request: Request, call_next):
             CORSMiddleware,
             allow_origins=settings.all_cors_origins,
             allow_credentials=True,
-            allow_methods=["*"],
-            allow_headers=["*"],
+            allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"],
+            allow_headers=[
+                "Authorization",
+                "Content-Type",
+                "Accept",
+                "Accept-Language",
+                "X-Requested-With",
+            ],
         )
+
+    # Reject requests whose Host header is not in the allowlist. In production
+    # this blocks Host-header injection / cache-poisoning; elsewhere the list
+    # resolves to ``["*"]`` so it is a no-op for local dev and tests.
+    app.add_middleware(TrustedHostMiddleware, allowed_hosts=settings.trusted_hosts)

app/models/user_activity.py

@@ -15,6 +15,14 @@
 from app.schemas.user_activity import ActivityStatus
 from app.utils import utc_now
 
+# Length caps for client-supplied audit fields, enforced at write time by the
+# logger (see ``app/use_cases/log_activity.py``). ``ip_address`` fits the
+# longest IPv6 textual form; ``user_agent`` is attacker-controlled so it is
+# bounded to stop oversized headers from bloating the audit log. Kept as a
+# write-time cap rather than a DB constraint to avoid a heavy column rewrite.
+IP_ADDRESS_MAX_LENGTH = 45
+USER_AGENT_MAX_LENGTH = 500
+
 
 class UserActivity(Base):
     """

app/repositories/admin/activity.py

@@ -10,6 +10,7 @@
 from app.models.user import User
 from app.models.user_activity import UserActivity
 from app.schemas.user_activity import ActivityStatus, ActivityType, ResourceType
+from app.utils.db_search import LIKE_ESCAPE_CHAR, ilike_contains
 
 
 def _filtered_activities_stmt(
@@ -30,12 +31,12 @@ def _filtered_activities_stmt(
     if user_search:
         # Match the acting user by name/email. ILIKE on raw columns so the
         # pg_trgm indexes on email/first_name/last_name can serve the query.
-        like = f"%{user_search}%"
+        like = ilike_contains(user_search)
         stmt = stmt.join(User, UserActivity.user_id == User.id).where(
             or_(
-                User.email.ilike(like),
-                User.first_name.ilike(like),
-                User.last_name.ilike(like),
+                User.email.ilike(like, escape=LIKE_ESCAPE_CHAR),
+                User.first_name.ilike(like, escape=LIKE_ESCAPE_CHAR),
+                User.last_name.ilike(like, escape=LIKE_ESCAPE_CHAR),
             )
         )
     if activity_type is not None:

app/repositories/admin/file.py

@@ -8,6 +8,7 @@
 
 from app.models.file import File
 from app.models.user import User
+from app.utils.db_search import LIKE_ESCAPE_CHAR, ilike_contains
 
 
 def _filtered_files_stmt(
@@ -23,12 +24,12 @@ def _filtered_files_stmt(
         # Inner-join the uploader and match on name or email. Files without an
         # uploader (uploaded_by_id IS NULL) are naturally excluded, which is
         # correct: they cannot match a person's name.
-        pattern = f"%{uploader}%"
+        pattern = ilike_contains(uploader)
         stmt = stmt.join(User, File.uploaded_by_id == User.id).where(
             or_(
-                User.first_name.ilike(pattern),
-                User.last_name.ilike(pattern),
-                User.email.ilike(pattern),
+                User.first_name.ilike(pattern, escape=LIKE_ESCAPE_CHAR),
+                User.last_name.ilike(pattern, escape=LIKE_ESCAPE_CHAR),
+                User.email.ilike(pattern, escape=LIKE_ESCAPE_CHAR),
             )
         )
     return stmt

app/repositories/admin/support.py

@@ -8,6 +8,7 @@
 
 from app.models.support import SupportTicket
 from app.models.user import User
+from app.utils.db_search import LIKE_ESCAPE_CHAR, ilike_contains
 
 
 def _filtered_tickets_stmt(
@@ -24,13 +25,13 @@ def _filtered_tickets_stmt(
     """
     stmt = select(SupportTicket)
     if search:
-        like = f"%{search}%"
+        like = ilike_contains(search)
         stmt = stmt.join(User, SupportTicket.user_id == User.id).where(
             or_(
-                SupportTicket.subject.ilike(like),
-                User.email.ilike(like),
-                User.first_name.ilike(like),
-                User.last_name.ilike(like),
+                SupportTicket.subject.ilike(like, escape=LIKE_ESCAPE_CHAR),
+                User.email.ilike(like, escape=LIKE_ESCAPE_CHAR),
+                User.first_name.ilike(like, escape=LIKE_ESCAPE_CHAR),
+                User.last_name.ilike(like, escape=LIKE_ESCAPE_CHAR),
             )
         )
     if status is not None:

app/repositories/admin/user.py

@@ -7,6 +7,7 @@
 
 from app.models.user import User
 from app.schemas.user import SystemRole
+from app.utils.db_search import LIKE_ESCAPE_CHAR, ilike_contains
 
 
 def _filtered_users_stmt(
@@ -22,12 +23,12 @@ def _filtered_users_stmt(
         # ``ILIKE`` on the raw columns so the ``pg_trgm`` GIN indexes on
         # email/first_name/last_name can actually serve the query. Wrapping
         # with ``func.lower(...)`` would defeat the index.
-        like = f"%{search}%"
+        like = ilike_contains(search)
         stmt = stmt.where(
             or_(
-                User.email.ilike(like),
-                User.first_name.ilike(like),
-                User.last_name.ilike(like),
+                User.email.ilike(like, escape=LIKE_ESCAPE_CHAR),
+                User.first_name.ilike(like, escape=LIKE_ESCAPE_CHAR),
+                User.last_name.ilike(like, escape=LIKE_ESCAPE_CHAR),
             )
         )
     if role is not None: