From 64d3882c16c171002f5bbe78961459858c1e742d Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Sun, 21 Jun 2026 21:03:20 +0200 Subject: [PATCH] M4-T01: add auth_login_throttle table + LoginThrottle model --- alembic_app/env.py | 1 + .../20260621_07_auth_login_throttle.py | 43 ++++ app/models/auth_throttle.py | 31 +++ docs/design/m4-login-hardening.md | 2 +- scripts/app_db_adopt.py | 2 +- tests/test_auth_throttle_model.py | 195 ++++++++++++++++++ 6 files changed, 272 insertions(+), 2 deletions(-) create mode 100644 alembic_app/versions/20260621_07_auth_login_throttle.py create mode 100644 app/models/auth_throttle.py create mode 100644 tests/test_auth_throttle_model.py diff --git a/alembic_app/env.py b/alembic_app/env.py index f198417..7760158 100644 --- a/alembic_app/env.py +++ b/alembic_app/env.py @@ -7,6 +7,7 @@ from app.config import get_settings from app.db import Base from app.models.config import AppConfigEntry # noqa: F401 from app.models.auth import AuthSession, AuthUser # noqa: F401 +from app.models.auth_throttle import LoginThrottle # noqa: F401 from app.models.public_ip import PublicIPHistory, PublicIPState # noqa: F401 from app.models.location import Location # noqa: F401 from app.models.poo import PooRecord # noqa: F401 diff --git a/alembic_app/versions/20260621_07_auth_login_throttle.py b/alembic_app/versions/20260621_07_auth_login_throttle.py new file mode 100644 index 0000000..28a8980 --- /dev/null +++ b/alembic_app/versions/20260621_07_auth_login_throttle.py @@ -0,0 +1,43 @@ +"""add auth_login_throttle table for exponential back-off throttling + +Revision ID: 20260621_07_auth_login_throttle +Revises: 20260611_06_merge_location_poo_tables +Create Date: 2026-06-21 00:00:00.000000 +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = "20260621_07_auth_login_throttle" +down_revision: Union[str, None] = "20260611_06_merge_location_poo_tables" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "auth_login_throttle", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("key", sa.String(length=255), nullable=False), + sa.Column("scope", sa.String(length=16), nullable=False), + sa.Column("failures", sa.Integer(), nullable=False), + sa.Column("first_failed_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("last_failed_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("next_allowed_at", sa.DateTime(timezone=True), nullable=True), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("scope", "key", name="uq_auth_login_throttle_scope_key"), + ) + op.create_index( + "ix_auth_login_throttle_scope_key", + "auth_login_throttle", + ["scope", "key"], + unique=False, + ) + + +def downgrade() -> None: + op.drop_index("ix_auth_login_throttle_scope_key", table_name="auth_login_throttle") + op.drop_table("auth_login_throttle") diff --git a/app/models/auth_throttle.py b/app/models/auth_throttle.py new file mode 100644 index 0000000..1364e74 --- /dev/null +++ b/app/models/auth_throttle.py @@ -0,0 +1,31 @@ +from datetime import datetime + +from sqlalchemy import DateTime, Index, Integer, String, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from app.db import Base + + +class LoginThrottle(Base): + """Tracks per-key login failure state for exponential back-off throttling. + + ``scope`` is either ``'ip'`` or ``'user'``; ``key`` is the IP address or + username string respectively. The pair ``(scope, key)`` is unique — one + row per tracked entity. A row is deleted on successful login (clear). + """ + + __tablename__ = "auth_login_throttle" + __table_args__ = ( + UniqueConstraint("scope", "key", name="uq_auth_login_throttle_scope_key"), + Index("ix_auth_login_throttle_scope_key", "scope", "key"), + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + key: Mapped[str] = mapped_column(String(255), nullable=False) + scope: Mapped[str] = mapped_column(String(16), nullable=False) # 'ip' | 'user' + failures: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + first_failed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + last_failed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + next_allowed_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) diff --git a/docs/design/m4-login-hardening.md b/docs/design/m4-login-hardening.md index 78c74a3..17abdd3 100644 --- a/docs/design/m4-login-hardening.md +++ b/docs/design/m4-login-hardening.md @@ -145,7 +145,7 @@ Phase B（TOTP，可选配） > 后端沿用校验闸门（`pytest`/`ruff`/改路由或 schema 则 `export_openapi` 重导出入库）。前端闸门见 §8。新增依赖（`pyotp`、`qrcode.react`）须同步 `requirements.in/.txt` 与 `frontend/package.json` 重新锁定。 ### M4-T01 — `auth_login_throttle` 表 + 模型 `[schema]` -- **Status**: `todo` · **Depends**: none +- **Status**: `done` · **Depends**: none - **Context**: 存按 IP / username 的失败退避状态。仅建 schema。 - **Files**: `create app/models/auth_throttle.py`（或并入 `app/models/auth.py`）；`create alembic_app/versions/_NN_auth_login_throttle.py`；`modify alembic_app/env.py`、`scripts/app_db_adopt.py`（baseline 常量）；`create tests/test_auth_throttle_model.py` - **Steps**: 模型 `LoginThrottle`（`id / key str / scope str('ip'|'user') / failures int / first_failed_at / last_failed_at / next_allowed_at`，`(scope,key)` unique 索引）；新 revision 建表 + 索引；更新 baseline 常量。 diff --git a/scripts/app_db_adopt.py b/scripts/app_db_adopt.py index acd0074..0b44d93 100644 --- a/scripts/app_db_adopt.py +++ b/scripts/app_db_adopt.py @@ -15,7 +15,7 @@ if str(PROJECT_ROOT) not in sys.path: from app.config import get_settings -APP_BASELINE_REVISION = "20260611_06_merge_location_poo_tables" +APP_BASELINE_REVISION = "20260621_07_auth_login_throttle" class AppDatabaseAdoptionError(RuntimeError): diff --git a/tests/test_auth_throttle_model.py b/tests/test_auth_throttle_model.py new file mode 100644 index 0000000..59eafbd --- /dev/null +++ b/tests/test_auth_throttle_model.py @@ -0,0 +1,195 @@ +"""Tests for M4-T01: LoginThrottle model and auth_login_throttle migration.""" +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path + +import pytest +from alembic import command +from alembic.config import Config +from sqlalchemy import create_engine, inspect +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from app.models.auth_throttle import LoginThrottle + + +def _make_app_alembic_config(database_url: str) -> Config: + config = Config("alembic_app.ini") + config.set_main_option("sqlalchemy.url", database_url) + return config + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def throttle_db(tmp_path: Path): + """Temporary SQLite DB upgraded to the current Alembic head.""" + db_path = tmp_path / "throttle_test.db" + db_url = f"sqlite:///{db_path}" + alembic_cfg = _make_app_alembic_config(db_url) + command.upgrade(alembic_cfg, "head") + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + yield engine + engine.dispose() + + +# --------------------------------------------------------------------------- +# Migration shape tests +# --------------------------------------------------------------------------- + + +def test_table_exists_after_upgrade(throttle_db): + """auth_login_throttle table must be present after upgrade to head.""" + inspector = inspect(throttle_db) + assert "auth_login_throttle" in inspector.get_table_names() + + +def test_table_has_expected_columns(throttle_db): + """All required columns must be present with correct nullability.""" + inspector = inspect(throttle_db) + columns = {col["name"]: col for col in inspector.get_columns("auth_login_throttle")} + + expected_non_nullable = {"id", "key", "scope", "failures", "first_failed_at", "last_failed_at"} + expected_nullable = {"next_allowed_at"} + + for col_name in expected_non_nullable: + assert col_name in columns, f"Missing column: {col_name}" + assert not columns[col_name]["nullable"], f"Column {col_name} should be NOT NULL" + + for col_name in expected_nullable: + assert col_name in columns, f"Missing column: {col_name}" + assert columns[col_name]["nullable"], f"Column {col_name} should be nullable" + + +def test_unique_constraint_scope_key(throttle_db): + """(scope, key) pair must be unique — inserting a duplicate must raise.""" + now = datetime.now(tz=timezone.utc) + with Session(throttle_db) as session: + session.add( + LoginThrottle( + key="192.168.1.1", + scope="ip", + failures=1, + first_failed_at=now, + last_failed_at=now, + ) + ) + session.commit() + + with Session(throttle_db) as session: + session.add( + LoginThrottle( + key="192.168.1.1", + scope="ip", + failures=2, + first_failed_at=now, + last_failed_at=now, + ) + ) + with pytest.raises(IntegrityError): + session.commit() + + +def test_same_key_different_scope_allowed(throttle_db): + """Same key string but different scopes are independent rows — both should insert.""" + now = datetime.now(tz=timezone.utc) + with Session(throttle_db) as session: + session.add( + LoginThrottle( + key="admin", + scope="ip", + failures=1, + first_failed_at=now, + last_failed_at=now, + ) + ) + session.add( + LoginThrottle( + key="admin", + scope="user", + failures=3, + first_failed_at=now, + last_failed_at=now, + ) + ) + session.commit() # must not raise + + with Session(throttle_db) as session: + rows = session.query(LoginThrottle).filter_by(key="admin").all() + assert len(rows) == 2 + scopes = {r.scope for r in rows} + assert scopes == {"ip", "user"} + + +def test_next_allowed_at_nullable(throttle_db): + """next_allowed_at is nullable — a row without it must persist fine.""" + now = datetime.now(tz=timezone.utc) + with Session(throttle_db) as session: + row = LoginThrottle( + key="10.0.0.1", + scope="ip", + failures=1, + first_failed_at=now, + last_failed_at=now, + next_allowed_at=None, + ) + session.add(row) + session.commit() + session.refresh(row) + assert row.next_allowed_at is None + + +def test_next_allowed_at_persists_when_set(throttle_db): + """next_allowed_at can be set and retrieved correctly.""" + now = datetime.now(tz=timezone.utc) + with Session(throttle_db) as session: + row = LoginThrottle( + key="10.0.0.2", + scope="ip", + failures=5, + first_failed_at=now, + last_failed_at=now, + next_allowed_at=now, + ) + session.add(row) + session.commit() + row_id = row.id + + with Session(throttle_db) as session: + fetched = session.get(LoginThrottle, row_id) + assert fetched is not None + assert fetched.next_allowed_at is not None + + +# --------------------------------------------------------------------------- +# Downgrade / upgrade reversibility test +# --------------------------------------------------------------------------- + + +def test_downgrade_removes_table(tmp_path: Path): + """downgrade -1 from head must cleanly drop auth_login_throttle.""" + db_path = tmp_path / "down_test.db" + db_url = f"sqlite:///{db_path}" + alembic_cfg = _make_app_alembic_config(db_url) + + command.upgrade(alembic_cfg, "head") + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + assert "auth_login_throttle" in inspect(engine).get_table_names() + engine.dispose() + + command.downgrade(alembic_cfg, "-1") + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + assert "auth_login_throttle" not in inspect(engine).get_table_names() + engine.dispose() + + +def test_index_exists_on_scope_key(throttle_db): + """ix_auth_login_throttle_scope_key composite index must exist.""" + inspector = inspect(throttle_db) + indexes = inspector.get_indexes("auth_login_throttle") + index_names = {idx["name"] for idx in indexes} + assert "ix_auth_login_throttle_scope_key" in index_names