busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

security: JWT aud/iss validation, token blacklist, encryption key crash-early

Browse Source 
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.
This commit is contained in:
busya
2026-05-26 15:08:55 +03:00
parent a9a453109c
commit 23719ecfbc
7 changed files with 212 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
backend/src/api/auth.py
View File

@@ -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"}

2
backend/src/core/auth/config.py
View File

@@ -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")

134
backend/src/core/auth/jwt.py
View File

@@ -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

49
backend/src/core/encryption_key.py
View File

@@ -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

13
backend/src/dependencies.py
View File

@@ -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:

16
backend/src/models/auth.py
View File

@@ -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]