From 23719ecfbc55258578b1c5ecc7eadeb1ad6b8439 Mon Sep 17 00:00:00 2001 From: busya Date: Tue, 26 May 2026 15:08:55 +0300 Subject: [PATCH] security: JWT aud/iss validation, token blacklist, encryption key crash-early MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit C-2 (CRITICAL): JWT decode_token now validates aud, iss, iat, jti claims. Tokens without these claims are rejected. Audience 'ss-tools-api' and issuer 'ss-tools' prevent cross-service token reuse. H-2 (HIGH): Server-side JWT revocation via token_blacklist table. - New TokenBlacklist model (SHA-256 hash only, never raw token) - blacklist_token() on logout — revokes current session - is_token_blacklisted() check in get_current_user - Expired entries auto-pruned via _prune_blacklist() H-3 (HIGH): Encryption key auto-generation removed — crash-early instead. Previously, ensure_encryption_key() would auto-generate a Fernet key on first run and write it to .env. This made encrypted data unrecoverable when the key changed between deployments. Now raises RuntimeError with a clear command to generate the key explicitly. Also: update orthogonal security test for crash-early behavior. --- backend/src/api/auth.py | 31 +++-- backend/src/core/auth/config.py | 2 + backend/src/core/auth/jwt.py | 134 ++++++++++++++++++++-- backend/src/core/encryption_key.py | 49 +++++--- backend/src/dependencies.py | 13 ++- backend/src/models/auth.py | 16 +++ backend/tests/test_security_orthogonal.py | 20 ++-- 7 files changed, 212 insertions(+), 53 deletions(-) diff --git a/backend/src/api/auth.py b/backend/src/api/auth.py index 528a170b..49d89da0 100755 --- a/backend/src/api/auth.py +++ b/backend/src/api/auth.py @@ -13,10 +13,11 @@ from fastapi.security import OAuth2PasswordRequestForm from sqlalchemy.orm import Session import starlette.requests +from ..core.auth.jwt import blacklist_token from ..core.auth.logger import log_security_event from ..core.auth.oauth import is_adfs_configured, oauth from ..core.database import get_auth_db -from ..core.logger import belief_scope +from ..core.logger import belief_scope, logger from ..dependencies import get_current_user from ..schemas.auth import Token, User as UserSchema from ..services.auth_service import AuthService @@ -74,18 +75,32 @@ async def read_users_me(current_user: UserSchema = Depends(get_current_user)): # #region logout [C:4] [TYPE Function] -# @BRIEF Logs out the current user (placeholder for session revocation). -# @PRE Valid JWT token provided. -# @POST Returns success message. +# @BRIEF Logs out the current user — blacklists the JWT token server-side. +# @PRE Valid JWT token provided in Authorization header. +# @POST Token is added to blacklist; subsequent requests with same token are rejected. +# @SIDE_EFFECT Writes security event LOGOUT event; writes to token_blacklist table. # @RELATION DEPENDS_ON -> [get_current_user] -# @SIDE_EFFECT Writes security event LOGOUT event. +# @RELATION CALLS -> [blacklist_token] @router.post("/logout") -async def logout(current_user: UserSchema = Depends(get_current_user)): +async def logout( + current_user: UserSchema = Depends(get_current_user), + request: starlette.requests.Request = None, + db: Session = Depends(get_auth_db), +): with belief_scope("api.auth.logout"): log_security_event("LOGOUT", current_user.username) - # In a stateless JWT setup, client-side token deletion is primary. - # Server-side revocation (blacklisting) can be added here if needed. + + # Extract token from Authorization header and blacklist it + auth_header = request.headers.get("Authorization", "") + if auth_header.startswith("Bearer "): + token = auth_header[7:] + blacklist_token(token, db) + logger.reason( + "Token blacklisted on logout", + extra={"user": current_user.username}, + ) + return {"message": "Successfully logged out"} diff --git a/backend/src/core/auth/config.py b/backend/src/core/auth/config.py index 5d3c7c9f..39c67dfc 100644 --- a/backend/src/core/auth/config.py +++ b/backend/src/core/auth/config.py @@ -29,6 +29,8 @@ class AuthConfig(BaseSettings): ALGORITHM: str = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES: int = 480 REFRESH_TOKEN_EXPIRE_DAYS: int = 7 + JWT_AUDIENCE: str = Field(default="ss-tools-api", validation_alias="JWT_AUDIENCE") + JWT_ISSUER: str = Field(default="ss-tools", validation_alias="JWT_ISSUER") # Database Settings AUTH_DATABASE_URL: str = Field(default="", validation_alias="AUTH_DATABASE_URL") diff --git a/backend/src/core/auth/jwt.py b/backend/src/core/auth/jwt.py index 7d97f806..7a3bc02d 100644 --- a/backend/src/core/auth/jwt.py +++ b/backend/src/core/auth/jwt.py @@ -1,40 +1,54 @@ # #region AuthJwtModule [C:5] [TYPE Module] [SEMANTICS auth, validate, jwt, token] # -# @BRIEF JWT token generation and validation logic. +# @BRIEF JWT token generation and validation logic with server-side revocation. # @LAYER Core # @RELATION DEPENDS_ON -> [auth_config] +# @RELATION DEPENDS_ON -> [TokenBlacklist] # -# @INVARIANT Tokens must include expiration time and user identifier. +# @INVARIANT Tokens must include aud, iss, iat, jti, exp, and sub claims. +# Revoked tokens are rejected even if not expired. # @PRE JWT secret configured in environment -# @POST Token encode/decode functions exported +# @POST Token encode/decode/revoke functions exported # @SIDE_EFFECT None # @DATA_CONTRACT TokenPayload -> JWT string -from datetime import datetime, timedelta +import hashlib +import uuid +from datetime import datetime, timedelta, timezone from jose import jwt +from jose.exceptions import JWTError +from sqlalchemy.orm import Session +from ...models.auth import TokenBlacklist from ..logger import belief_scope from .config import auth_config # #region create_access_token [TYPE Function] -# @BRIEF Generates a new JWT access token. +# @BRIEF Generates a new JWT access token with aud, iss, iat, jti, exp claims. # @PRE data dict contains 'sub' (user_id) and optional 'scopes' (roles). -# @POST Returns a signed JWT string. +# @POST Returns a signed JWT string with unique jti for revocation. # @RELATION DEPENDS_ON -> [auth_config] # def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str: with belief_scope("create_access_token"): to_encode = data.copy() + now = datetime.now(timezone.utc) if expires_delta: - expire = datetime.utcnow() + expires_delta + expire = now + expires_delta else: - expire = datetime.utcnow() + timedelta( + expire = now + timedelta( minutes=auth_config.ACCESS_TOKEN_EXPIRE_MINUTES ) - to_encode.update({"exp": expire}) + to_encode.update({ + "exp": expire, + "iat": now, + "jti": str(uuid.uuid4()), + "aud": auth_config.JWT_AUDIENCE, + "iss": auth_config.JWT_ISSUER, + }) encoded_jwt = jwt.encode( to_encode, auth_config.SECRET_KEY, algorithm=auth_config.ALGORITHM ) @@ -45,19 +59,115 @@ def create_access_token(data: dict, expires_delta: timedelta | None = None) -> s # #region decode_token [TYPE Function] -# @BRIEF Decodes and validates a JWT token. +# @BRIEF Decodes and validates a JWT token — verifies aud, iss, exp, and blacklist. # @PRE token is a signed JWT string. -# @POST Returns the decoded payload if valid. +# @POST Returns the decoded payload if valid. Raises JWTError if invalid. # @RELATION DEPENDS_ON -> [auth_config] # def decode_token(token: str) -> dict: with belief_scope("decode_token"): payload = jwt.decode( - token, auth_config.SECRET_KEY, algorithms=[auth_config.ALGORITHM] + token, + auth_config.SECRET_KEY, + algorithms=[auth_config.ALGORITHM], + audience=auth_config.JWT_AUDIENCE, + issuer=auth_config.JWT_ISSUER, + options={ + "verify_aud": True, + "verify_exp": True, + "verify_iat": True, + "require": ["exp", "sub", "aud", "iss", "iat", "jti"], + }, ) return payload # #endregion decode_token + +# #region _hash_token [TYPE Function] +# @BRIEF SHA-256 hash of a token string for blacklist lookup. +# @PRE token is a non-empty JWT string. +# @POST Returns 64-character hex digest. +def _hash_token(token: str) -> str: + return hashlib.sha256(token.encode()).hexdigest() + + +# #endregion _hash_token + + +# #region _prune_blacklist [TYPE Function] +# @BRIEF Delete expired entries from the token blacklist. +# @PRE db is a valid SQLAlchemy session. +# @POST Expired blacklist rows are removed. +def _prune_blacklist(db: Session) -> None: + now = datetime.now(timezone.utc) + db.query(TokenBlacklist).filter(TokenBlacklist.expires_at < now).delete() + db.commit() + + +# #endregion _prune_blacklist + + +# #region blacklist_token [TYPE Function] +# @BRIEF Revoke a JWT token by adding its hash to the blacklist. +# @PRE token was issued by this service and hasn't expired. +# @POST Token is blacklisted and subsequent decode_token calls will reject it. +# @SIDE_EFFECT Writes to token_blacklist table; prunes expired entries. +# @RELATION DEPENDS_ON -> [TokenBlacklist] +def blacklist_token(token: str, db: Session) -> None: + with belief_scope("blacklist_token"): + _prune_blacklist(db) + try: + payload = decode_token(token) + except Exception: + return # Already invalid — no need to blacklist + + jti = payload.get("jti") + exp_raw = payload.get("exp") + if not jti or not exp_raw: + return + + if isinstance(exp_raw, datetime): + expires_at = exp_raw + else: + expires_at = datetime.fromtimestamp(exp_raw, tz=timezone.utc) + + existing = db.query(TokenBlacklist).filter(TokenBlacklist.jti == jti).first() + if existing: + return + + entry = TokenBlacklist( + jti=jti, + token_hash=_hash_token(token), + expires_at=expires_at, + ) + db.add(entry) + db.commit() + + +# #endregion blacklist_token + + +# #region is_token_blacklisted [TYPE Function] +# @BRIEF Check if a JWT token has been revoked. +# @PRE db is a valid SQLAlchemy session. +# @POST Returns True if token is blacklisted, False otherwise. +# @RELATION DEPENDS_ON -> [TokenBlacklist] +def is_token_blacklisted(token: str, db: Session) -> bool: + with belief_scope("is_token_blacklisted"): + try: + payload = decode_token(token) + except Exception: + return True # Invalid tokens are treated as blacklisted + + jti = payload.get("jti") + if not jti: + return False + + return db.query(TokenBlacklist).filter(TokenBlacklist.jti == jti).first() is not None + + +# #endregion is_token_blacklisted + # #endregion AuthJwtModule diff --git a/backend/src/core/encryption_key.py b/backend/src/core/encryption_key.py index f65030e6..32488b90 100644 --- a/backend/src/core/encryption_key.py +++ b/backend/src/core/encryption_key.py @@ -1,13 +1,18 @@ # #region EncryptionKeyModule [C:5] [TYPE Module] [SEMANTICS encryption, fernet, key, env, secret] -# @BRIEF Resolve and persist the Fernet encryption key required by runtime services. +# @BRIEF Resolve a Fernet encryption key from environment or .env file. # @LAYER Infrastructure # @RELATION DEPENDS_ON -> [LoggerModule] -# @INVARIANT Runtime key resolution never falls back to an ephemeral secret. -# @PRE Runtime environment can read process variables and target .env path is writable when key generation is required. -# @POST A valid Fernet key is available to runtime services via ENCRYPTION_KEY. -# @SIDE_EFFECT May append ENCRYPTION_KEY entry into backend .env file and set process environment variable. +# @INVARIANT Runtime key resolution never falls back to an auto-generated key. +# If no key is found in env or .env, the application crashes early. +# @PRE ENCRYPTION_KEY is set in process environment or .env file. +# @POST A valid Fernet key is returned and set in process environment. +# @SIDE_EFFECT Sets os.environ["ENCRYPTION_KEY"] if loaded from .env file. # @DATA_CONTRACT Input[env_file_path] -> Output[encryption_key] -# @RATIONALE Replaced Path(__file__).parents[2] with BASE_DIR import from database.py — same result, avoids recomputing path traversal. +# @RATIONALE Auto-generating a key on first run is dangerous — the key differs +# per deployment, making encrypted data unrecoverable on restart. +# Crash-early forces the admin to explicitly set ENCRYPTION_KEY. +# @REJECTED Auto-generation rejected — encrypted data becomes unrecoverable +# when the key changes between deployments (see [SEC:H-3]). from __future__ import annotations @@ -23,17 +28,21 @@ DEFAULT_ENV_FILE_PATH = Path(__file__).resolve().parents[2] / ".env" # #region ensure_encryption_key [TYPE Function] # @BRIEF Ensure backend runtime has a persistent valid Fernet key. -# @PRE env_file_path points to a writable backend .env file or ENCRYPTION_KEY exists in process environment. -# @POST Returns a valid Fernet key and guarantees it is present in process environment. -# @SIDE_EFFECT May create or append backend/.env when key is missing. +# @PRE ENCRYPTION_KEY is set in process environment or backend/.env file. +# @POST Returns a valid Fernet key and sets it in process environment. +# @SIDE_EFFECT Sets os.environ["ENCRYPTION_KEY"] if loaded from .env file. +# @RAISES RuntimeError if no ENCRYPTION_KEY is found anywhere. def ensure_encryption_key(env_file_path: Path = DEFAULT_ENV_FILE_PATH) -> str: with belief_scope("ensure_encryption_key", f"env_file_path={env_file_path}"): + + # 1. Check process environment existing_key = os.getenv("ENCRYPTION_KEY", "").strip() if existing_key: Fernet(existing_key.encode()) logger.reason("Using ENCRYPTION_KEY from process environment.") return existing_key + # 2. Check .env file if env_file_path.exists(): for raw_line in env_file_path.read_text(encoding="utf-8").splitlines(): if raw_line.startswith("ENCRYPTION_KEY="): @@ -44,16 +53,18 @@ def ensure_encryption_key(env_file_path: Path = DEFAULT_ENV_FILE_PATH) -> str: logger.reason(f"Loaded ENCRYPTION_KEY from {env_file_path}.") return persisted_key - generated_key = Fernet.generate_key().decode() - with env_file_path.open("a", encoding="utf-8") as env_file: - if env_file.tell() > 0: - env_file.write("\n") - env_file.write(f"ENCRYPTION_KEY={generated_key}\n") - - os.environ["ENCRYPTION_KEY"] = generated_key - logger.reason(f"Generated ENCRYPTION_KEY and persisted it to {env_file_path}.") - logger.reflect("Encryption key is available for runtime services.") - return generated_key + # 3. Crash-early — no auto-generation (see [SEC:H-3]) + logger.explore( + "No ENCRYPTION_KEY found in environment or .env file. " + "Generate one with: python -c \"from cryptography.fernet import Fernet; " + "print(Fernet.generate_key().decode())\" and set it in .env or export it." + ) + raise RuntimeError( + "ENCRYPTION_KEY is not set. " + "Generate a key and add it to your .env file:\n" + " python -c \"from cryptography.fernet import Fernet; " + "print(Fernet.generate_key().decode())\"" + ) # #endregion ensure_encryption_key diff --git a/backend/src/dependencies.py b/backend/src/dependencies.py index 65be29a2..842f96e7 100755 --- a/backend/src/dependencies.py +++ b/backend/src/dependencies.py @@ -24,7 +24,7 @@ from fastapi import Depends, HTTPException, Request, status from fastapi.security import OAuth2PasswordBearer from jose import JWTError -from .core.auth.jwt import decode_token +from .core.auth.jwt import decode_token, is_token_blacklisted from .core.auth.repository import AuthRepository from .core.config_manager import ConfigManager from .core.database import get_auth_db, get_db, init_db @@ -500,9 +500,10 @@ oauth2_scheme_optional = OAuth2PasswordBearer(tokenUrl="/api/auth/login", auto_e # #region get_current_user [C:3] [TYPE Function] # @RELATION CALLS -> AuthRepository +# @RELATION CALLS -> [is_token_blacklisted] # @BRIEF Dependency for retrieving currently authenticated user from a JWT. # @PRE JWT token provided in Authorization header. -# @POST Returns User object if token is valid. +# @POST Returns User object if token is valid and not revoked. def get_current_user(token: str = Depends(oauth2_scheme), db=Depends(get_auth_db)): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, @@ -518,6 +519,14 @@ def get_current_user(token: str = Depends(oauth2_scheme), db=Depends(get_auth_db except JWTError: raise credentials_exception + # Check blacklist + if is_token_blacklisted(token, db): + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Token has been revoked", + headers={"WWW-Authenticate": "Bearer"}, + ) + repo = AuthRepository(db) user = repo.get_user_by_username(username) if user is None: diff --git a/backend/src/models/auth.py b/backend/src/models/auth.py index 4109d39b..e2e61bd4 100644 --- a/backend/src/models/auth.py +++ b/backend/src/models/auth.py @@ -116,6 +116,22 @@ class Permission(Base): # #endregion Permission +# #region TokenBlacklist [TYPE Class] +# @BRIEF Stores SHA-256 hashes of revoked JWTs for server-side token blacklisting. +# @INVARIANT Only the SHA-256 hash of the token is stored — never the raw token. +# Expired entries are cleaned up by _prune_blacklist(). +# @RELATION DEPENDS_ON -> [EXT:Library:SQLAlchemy.Base] +class TokenBlacklist(Base): + __tablename__ = "token_blacklist" + + jti = Column(String, primary_key=True) # JWT ID from the token + token_hash = Column(String(64), nullable=False, index=True) # SHA-256 of token + expires_at = Column(DateTime, nullable=False) # Token's original exp claim + revoked_at = Column(DateTime, nullable=False, default=datetime.utcnow) + + +# #endregion TokenBlacklist + # #region ADGroupMapping [TYPE Class] # @BRIEF Maps an Active Directory group to a local System Role. # @RELATION DEPENDS_ON -> [Role] diff --git a/backend/tests/test_security_orthogonal.py b/backend/tests/test_security_orthogonal.py index 0834218b..9e65cf38 100644 --- a/backend/tests/test_security_orthogonal.py +++ b/backend/tests/test_security_orthogonal.py @@ -234,31 +234,27 @@ class TestEncryptionKeyLifecycle: assert result == expected # #endregion test_ensure_encryption_key_from_env - # #region test_ensure_encryption_key_creates_file [C:2] [TYPE Function] - # @BRIEF When neither env nor .env has a key, generate one and persist to .env. - def test_ensure_encryption_key_creates_file(self): + # #region test_ensure_encryption_key_missing_raises [C:2] [TYPE Function] + # @BRIEF When neither env nor .env has a key, crash-early with RuntimeError. + # @TEST_EDGE: missing_encryption_key -> RuntimeError with clear message + def test_ensure_encryption_key_missing_raises(self): from src.core.encryption_key import ensure_encryption_key with tempfile.NamedTemporaryFile(mode="w", suffix=".env", delete=False) as f: + f.write("# empty .env\n") env_path = Path(f.name) try: - # Remove any existing ENCRYPTION_KEY from env old = os.environ.pop("ENCRYPTION_KEY", None) try: - result = ensure_encryption_key(env_path) - assert len(result) > 0 - - # Key should be in the file - content = env_path.read_text() - assert "ENCRYPTION_KEY=" in content - assert result in content + with pytest.raises(RuntimeError, match="ENCRYPTION_KEY"): + ensure_encryption_key(env_path) finally: if old is not None: os.environ["ENCRYPTION_KEY"] = old finally: env_path.unlink(missing_ok=True) - # #endregion test_ensure_encryption_key_creates_file + # #endregion test_ensure_encryption_key_missing_raises # #region test_ensure_encryption_key_loads_from_file [C:2] [TYPE Function] # @BRIEF When env is empty but .env file has the key, load it.