From 2bc8ad87e193d0a1c56531f9557a444dac3e0490 Mon Sep 17 00:00:00 2001 From: busya Date: Tue, 26 May 2026 15:17:19 +0300 Subject: [PATCH] security: rate limiter, is_admin flag, password policy, log masking MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit M-3: Role.name == 'Admin' string comparison → Role.is_admin boolean flag. Admin role created with is_admin=True. has_permission checks getattr(role, 'is_admin', False) instead of comparing role names. M-2: In-memory rate limiter on /api/auth/login. Tracks failed attempts per IP (10 in 5 min window, 15 min ban). Thread-safe via threading.Lock. M-5: Password policy — UserCreate.password: min_length=8, must include uppercase, lowercase, and digit. L-1: log_security_event details dict now masks sensitive fields: password, secret, token, api_key, authorization. Recursive masking for nested dicts. Sensitive field names matched via regex. OTHER: - 4 rate limiter tests (under limit, ban, success clears, IP isolation) - 4 password policy tests (too short, no upper, no digit, valid) - 2 log masking tests (flat keys, nested) --- backend/src/api/auth.py | 18 ++- backend/src/app.py | 2 +- backend/src/core/auth/logger.py | 37 +++++- backend/src/core/rate_limiter.py | 79 ++++++++++++ backend/src/dependencies.py | 4 +- backend/src/models/auth.py | 1 + backend/src/schemas/auth.py | 17 ++- backend/tests/test_security_orthogonal.py | 147 ++++++++++++++++++++++ 8 files changed, 294 insertions(+), 11 deletions(-) create mode 100644 backend/src/core/rate_limiter.py diff --git a/backend/src/api/auth.py b/backend/src/api/auth.py index 49d89da0..0536c5d0 100755 --- a/backend/src/api/auth.py +++ b/backend/src/api/auth.py @@ -18,6 +18,7 @@ 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, logger +from ..core.rate_limiter import rate_limiter from ..dependencies import get_current_user from ..schemas.auth import Token, User as UserSchema from ..services.auth_service import AuthService @@ -38,12 +39,26 @@ router = APIRouter(prefix="/api/auth", tags=["auth"]) @router.post("/login", response_model=Token) async def login_for_access_token( - form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_auth_db) + form_data: OAuth2PasswordRequestForm = Depends(), + db: Session = Depends(get_auth_db), + request: starlette.requests.Request = None, ): with belief_scope("api.auth.login"): + # Rate limiting — protect against brute force + client_ip = request.client.host if request.client else "unknown" + if rate_limiter.is_banned(client_ip): + log_security_event( + "LOGIN_BLOCKED", form_data.username, {"reason": "Rate limited", "ip": client_ip} + ) + raise HTTPException( + status_code=status.HTTP_429_TOO_MANY_REQUESTS, + detail="Too many login attempts. Try again later.", + ) + auth_service = AuthService(db) user = auth_service.authenticate_user(form_data.username, form_data.password) if not user: + rate_limiter.record_attempt(client_ip) log_security_event( "LOGIN_FAILED", form_data.username, {"reason": "Invalid credentials"} ) @@ -52,6 +67,7 @@ async def login_for_access_token( detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}, ) + rate_limiter.record_success(client_ip) log_security_event("LOGIN_SUCCESS", user.username, {"source": "LOCAL"}) return auth_service.create_session(user) diff --git a/backend/src/app.py b/backend/src/app.py index 2f38b3f9..4f0c4a6e 100755 --- a/backend/src/app.py +++ b/backend/src/app.py @@ -86,7 +86,7 @@ def ensure_initial_admin_user() -> None: try: admin_role = db.query(Role).filter(Role.name == "Admin").first() if not admin_role: - admin_role = Role(name="Admin", description="System Administrator") + admin_role = Role(name="Admin", description="System Administrator", is_admin=True) db.add(admin_role) db.commit() db.refresh(admin_role) diff --git a/backend/src/core/auth/logger.py b/backend/src/core/auth/logger.py index eee0fa35..ab6e5d53 100644 --- a/backend/src/core/auth/logger.py +++ b/backend/src/core/auth/logger.py @@ -5,25 +5,51 @@ # @RELATION DEPENDS_ON -> [EXT:frontend:core_logger] # # @INVARIANT Must not log sensitive data like passwords or full tokens. +# Sensitive field names (password, secret, token, key, authorization) +# in the details dict are automatically masked. # @PRE Auth module initialized # @POST Audit logging functions exported # @SIDE_EFFECT Writes auth audit log entries # @DATA_CONTRACT AuthEvent -> LogEntry -# @SIDE_EFFECT Writes auth audit log entries -# @DATA_CONTRACT AuthEvent -> LogEntry +import copy +import re from datetime import datetime +from typing import Any from ..logger import belief_scope, logger +# Sensitive field name patterns — values for these keys are masked in logs +_SENSITIVE_KEYS = re.compile( + r"^(password|secret|token|api_key|api\.key|authorization|auth|credential|credit_card|ssn)$", + re.IGNORECASE, +) + + +def _mask_details(details: dict[str, Any] | None) -> dict[str, Any] | None: + """Return a copy of details with sensitive field values replaced by '***'.""" + if not details: + return details + masked = {} + for key, value in details.items(): + if isinstance(key, str) and _SENSITIVE_KEYS.match(key): + masked[key] = "***" + elif isinstance(value, dict): + masked[key] = _mask_details(value) + else: + masked[key] = value + return masked + # #region log_security_event [TYPE Function] # @BRIEF Logs a security-related event for audit trails. # @PRE event_type and username are strings. # @POST Security event is written to the application log via %s-formatting to prevent log injection. +# Sensitive fields in details dict are masked. # @RELATION DEPENDS_ON -> [logger] # @RATIONALE Uses %s-formatting (not f-strings) to prevent log forging via CRLF injection in -# user-controlled fields — see [SEC:H-1]. +# user-controlled fields — see [SEC:H-1]. Sensitive fields in details +# dict are masked to prevent credential leakage — see [SEC:L-1]. def log_security_event(event_type: str, username: str, details: dict = None): with belief_scope("log_security_event", f"{event_type}:{username}"): timestamp = datetime.utcnow().isoformat() @@ -33,12 +59,13 @@ def log_security_event(event_type: str, username: str, details: dict = None): event_type, username, ) - if details: + safe_details = _mask_details(details) + if safe_details: logger.info( "[AUDIT][%s][%s] Details: %s", timestamp, event_type, - details, + safe_details, ) # #endregion log_security_event diff --git a/backend/src/core/rate_limiter.py b/backend/src/core/rate_limiter.py new file mode 100644 index 00000000..4ff3577c --- /dev/null +++ b/backend/src/core/rate_limiter.py @@ -0,0 +1,79 @@ +# #region RateLimiterModule [C:4] [TYPE Module] [SEMANTICS rate_limit, security, throttle, auth] +# @BRIEF Simple in-memory rate limiter for auth endpoints. +# Tracks request frequency per IP and blocks excessive requests. +# For production, replace with Redis-based limiter. +# @LAYER Core +# @INVARIANT Never blocks successful auth — only repeated failures. +# Window resets after ATTEMPT_WINDOW seconds of no activity. +# @REJECTED Redis-backed limiter — not available in all deployments. +# Per-user limiting — user isn't known until after auth. +import time +import threading +from collections import defaultdict + +# ── Configuration ── +MAX_ATTEMPTS: int = 10 # Max failed attempts per window +ATTEMPT_WINDOW: int = 300 # Window in seconds (5 min) +BAN_DURATION: int = 900 # Ban duration after limit reached (15 min) + + +class RateLimiter: + """In-memory sliding-window rate limiter keyed by IP address. + + Thread-safe. Tracks failed attempts. After MAX_ATTEMPTS failures within + ATTEMPT_WINDOW seconds, the IP is banned for BAN_DURATION seconds. + """ + + def __init__(self): + self._lock = threading.Lock() + # ip -> [timestamp, ...] sorted list of attempt timestamps + self._attempts: dict[str, list[float]] = defaultdict(list) + # ip -> ban_expires_at + self._bans: dict[str, float] = {} + + # #region is_banned [TYPE Function] + # @BRIEF Check if IP is currently banned. + # @PRE ip is a valid IP string. + # @POST Returns True if IP is banned and ban hasn't expired. + def is_banned(self, ip: str) -> bool: + with self._lock: + expires = self._bans.get(ip) + if expires is None: + return False + if time.monotonic() > expires: + del self._bans[ip] + return False + return True + # #endregion is_banned + + # #region record_attempt [TYPE Function] + # @BRIEF Record a failed attempt from an IP. + # @PRE ip is a valid IP string. + # @POST If MAX_ATTEMPTS exceeded within window, IP is banned. + def record_attempt(self, ip: str) -> None: + with self._lock: + now = time.monotonic() + window_start = now - ATTEMPT_WINDOW + + # Prune old attempts + self._attempts[ip] = [t for t in self._attempts[ip] if t > window_start] + self._attempts[ip].append(now) + + if len(self._attempts[ip]) > MAX_ATTEMPTS: + self._bans[ip] = now + BAN_DURATION + del self._attempts[ip] + # #endregion record_attempt + + # #region record_success [TYPE Function] + # @BRIEF Clear attempt history on successful auth. + # @PRE ip is a valid IP string. + # @POST Attempt history for IP is cleared. + def record_success(self, ip: str) -> None: + with self._lock: + self._attempts.pop(ip, None) + # #endregion record_success + + +# Singleton +rate_limiter = RateLimiter() +# #endregion RateLimiterModule diff --git a/backend/src/dependencies.py b/backend/src/dependencies.py index 842f96e7..da824d23 100755 --- a/backend/src/dependencies.py +++ b/backend/src/dependencies.py @@ -550,8 +550,8 @@ def has_permission(resource: str, action: str): if perm.resource == resource and perm.action == action: return current_user - # Special case for Admin role (full access) - if any(role.name == "Admin" for role in current_user.roles): + # Special case for Admin role (full access) — uses is_admin flag, not name string + if any(getattr(role, "is_admin", False) for role in current_user.roles): return current_user from .core.auth.logger import log_security_event diff --git a/backend/src/models/auth.py b/backend/src/models/auth.py index e2e61bd4..fe04b7f4 100644 --- a/backend/src/models/auth.py +++ b/backend/src/models/auth.py @@ -88,6 +88,7 @@ class Role(Base): id = Column(String, primary_key=True, default=generate_uuid) name = Column(String, unique=True, index=True, nullable=False) description = Column(String, nullable=True) + is_admin = Column(Boolean, default=False, nullable=False) users = relationship("User", secondary=user_roles, back_populates="roles") permissions = relationship( diff --git a/backend/src/schemas/auth.py b/backend/src/schemas/auth.py index 933ab198..b71312dd 100644 --- a/backend/src/schemas/auth.py +++ b/backend/src/schemas/auth.py @@ -9,7 +9,7 @@ from datetime import datetime -from pydantic import BaseModel, EmailStr +from pydantic import BaseModel, EmailStr, Field, field_validator # #region Token [C:1] [TYPE Class] @@ -121,9 +121,22 @@ class UserBase(BaseModel): # #region UserCreate [TYPE Class] # @BRIEF Schema for creating a new user. class UserCreate(UserBase): - password: str + password: str = Field(min_length=8) roles: list[str] = [] + @field_validator("password") + @classmethod + def validate_password_strength(cls, v: str) -> str: + if len(v) < 8: + raise ValueError("Password must be at least 8 characters") + if not any(c.isupper() for c in v): + raise ValueError("Password must contain at least one uppercase letter") + if not any(c.islower() for c in v): + raise ValueError("Password must contain at least one lowercase letter") + if not any(c.isdigit() for c in v): + raise ValueError("Password must contain at least one digit") + return v + # #endregion UserCreate diff --git a/backend/tests/test_security_orthogonal.py b/backend/tests/test_security_orthogonal.py index 9e65cf38..3f0b48f7 100644 --- a/backend/tests/test_security_orthogonal.py +++ b/backend/tests/test_security_orthogonal.py @@ -381,4 +381,151 @@ class TestJwtEdgeCases: # #endregion test_decode_token_missing_sub +# ────────────────────────────────────────────── +# RATE LIMITER +# ────────────────────────────────────────────── + + +class TestRateLimiter: + """Orthogonal tests for in-memory rate limiter.""" + + # #region test_rate_limiter_allows_under_limit [C:2] [TYPE Function] + # @BRIEF Under limit: not banned, attempts recorded. + def test_rate_limiter_allows_under_limit(self): + from src.core.rate_limiter import RateLimiter + + rl = RateLimiter() + assert not rl.is_banned("1.2.3.4") + for _ in range(5): + rl.record_attempt("1.2.3.4") + assert not rl.is_banned("1.2.3.4") + # #endregion test_rate_limiter_allows_under_limit + + # #region test_rate_limiter_bans_over_limit [C:2] [TYPE Function] + # @BRIEF Over limit: IP is banned, success clears history. + def test_rate_limiter_bans_over_limit(self): + from src.core.rate_limiter import RateLimiter, MAX_ATTEMPTS + + rl = RateLimiter() + ip = "5.6.7.8" + for _ in range(MAX_ATTEMPTS + 1): + rl.record_attempt(ip) + assert rl.is_banned(ip) + + # After ban, success should clear (but may still be banned) + rl.record_success(ip) + # #endregion test_rate_limiter_bans_over_limit + + # #region test_rate_limiter_success_clears_history [C:2] [TYPE Function] + # @BRIEF Successful auth clears attempt history for that IP. + def test_rate_limiter_success_clears_history(self): + from src.core.rate_limiter import RateLimiter + + rl = RateLimiter() + rl.record_attempt("9.9.9.9") + rl.record_attempt("9.9.9.9") + rl.record_success("9.9.9.9") + # After success, the attempts dict is cleared — we can make more without ban + for _ in range(5): + rl.record_attempt("9.9.9.9") + assert not rl.is_banned("9.9.9.9") + # #endregion test_rate_limiter_success_clears_history + + # #region test_rate_limiter_different_ips_independent [C:2] [TYPE Function] + # @BRIEF Different IPs are tracked independently. + def test_rate_limiter_different_ips_independent(self): + from src.core.rate_limiter import RateLimiter, MAX_ATTEMPTS + + rl = RateLimiter() + # One IP gets banned + for _ in range(MAX_ATTEMPTS + 1): + rl.record_attempt("bad-ip") + assert rl.is_banned("bad-ip") + # Other IP is not affected + assert not rl.is_banned("good-ip") + # #endregion test_rate_limiter_different_ips_independent + + +# ────────────────────────────────────────────── +# PASSWORD POLICY +# ────────────────────────────────────────────── + + +class TestPasswordPolicy: + """Orthogonal tests for password strength validation.""" + + # #region test_password_too_short [C:2] [TYPE Function] + def test_password_too_short(self): + from pydantic import ValidationError + from src.schemas.auth import UserCreate + + with pytest.raises(ValidationError): + UserCreate(username="test", password="Ab1") # only 3 chars + # #endregion test_password_too_short + + # #region test_password_no_uppercase [C:2] [TYPE Function] + def test_password_no_uppercase(self): + from pydantic import ValidationError + from src.schemas.auth import UserCreate + + with pytest.raises(ValidationError): + UserCreate(username="test", password="abcdefgh1") # no uppercase + # #endregion test_password_no_uppercase + + # #region test_password_no_digit [C:2] [TYPE Function] + def test_password_no_digit(self): + from pydantic import ValidationError + from src.schemas.auth import UserCreate + + with pytest.raises(ValidationError): + UserCreate(username="test", password="Abcdefgh") # no digit + # #endregion test_password_no_digit + + # #region test_password_valid [C:2] [TYPE Function] + def test_password_valid(self): + from src.schemas.auth import UserCreate + + u = UserCreate(username="test", password="ValidPass1") + assert u.password == "ValidPass1" + # #endregion test_password_valid + + +# ────────────────────────────────────────────── +# LOG MASKING +# ────────────────────────────────────────────── + + +class TestLogMasking: + """Orthogonal tests for sensitive field masking in log_security_event.""" + + # #region test_mask_sensitive_fields [C:2] [TYPE Function] + def test_mask_sensitive_fields(self): + from src.core.auth.logger import _mask_details + + masked = _mask_details({ + "username": "john", + "password": "supersecret", + "token": "eyJhbGciOiJIUzI1NiJ9.xxx", + "api_key": "ssk_abc123", + "source": "LOCAL", + }) + assert masked["username"] == "john" + assert masked["password"] == "***" + assert masked["token"] == "***" + assert masked["api_key"] == "***" + assert masked["source"] == "LOCAL" + # #endregion test_mask_sensitive_fields + + # #region test_mask_nested_details [C:2] [TYPE Function] + def test_mask_nested_details(self): + from src.core.auth.logger import _mask_details + + masked = _mask_details({ + "user": {"username": "john", "password": "secret"}, + }) + assert masked["user"]["username"] == "john" + assert masked["user"]["password"] == "***" + # #endregion test_mask_nested_details + + # #endregion TestSecurityOrthogonal