Merge pull request #103 from amar-python/feat/m2a2-login-lockout

feat(auth): lock accounts after repeated failed sign-ins (#102)

Author: amar-python

alembic/versions/0005_login_lockout.py

@@ -0,0 +1,41 @@
+"""Lock accounts after repeated failed sign-ins (#102, BR-36).
+
+Shared-only, like 0004: users live in the control schema, not in each tenant.
+
+Revision ID: 0005
+Revises: 0004
+Create Date: 2026-10-26
+"""
+
+from alembic import context, op
+
+from api.config import settings
+
+revision = "0005"
+down_revision = "0004"
+branch_labels = None
+depends_on = None
+
+
+def _skip() -> bool:
+    return context.config.attributes.get("scope") == "tenant"
+
+
+def _users() -> str:
+    return op.get_bind().dialect.identifier_preparer.quote(settings.CONTROL_SCHEMA) + ".users"
+
+
+def upgrade() -> None:
+    if _skip():
+        return
+    op.execute(f"""ALTER TABLE {_users()}
+        ADD COLUMN failed_logins INTEGER NOT NULL DEFAULT 0,
+        ADD COLUMN first_failed_at TIMESTAMPTZ,
+        ADD COLUMN locked_until TIMESTAMPTZ""")
+
+
+def downgrade() -> None:
+    if _skip():
+        return
+    op.execute(f"""ALTER TABLE {_users()}
+        DROP COLUMN failed_logins, DROP COLUMN first_failed_at, DROP COLUMN locked_until""")

api/config.py

@@ -16,6 +16,10 @@ class Settings:
     CONTROL_SCHEMA: str = os.environ.get("CONTROL_SCHEMA", "control")
     # Lifetime of a sign-in session, in hours.
     SESSION_TTL_HOURS: int = int(os.environ.get("SESSION_TTL_HOURS", "12"))
+    # #102: lock an account after this many failed sign-ins within the window,
+    # for the same number of minutes.
+    LOGIN_MAX_FAILURES: int = int(os.environ.get("LOGIN_MAX_FAILURES", "10"))
+    LOGIN_LOCKOUT_MINUTES: int = int(os.environ.get("LOGIN_LOCKOUT_MINUTES", "15"))
 
     # Schema holding the fixed T&E tables (deploy_all.sh dev creates te_dev).
     TE_SCHEMA: str = os.environ.get("TE_SCHEMA", "te_dev")

api/services/accounts.py

@@ -68,19 +68,25 @@ def add_membership(cur, user_id: int, tenant_id: int, role: str) -> None:
 def login(cur, tenant_slug: str, username: str, password: str) -> dict | None:
     """Return a new session for an active member of the tenant, else None.
 
-    The caller commits. Wrong tenant, unknown user, wrong password and inactive
-    user all return None, so a caller cannot tell which one failed.
+    The caller commits, including on failure so failed attempts are counted.
+    Wrong tenant, unknown user, wrong password, inactive user and locked
+    account (#102) all return None, so a caller cannot tell which one failed.
     """
     cur.execute(_q(
-        "SELECT u.id, u.password_hash, u.is_active, t.id, m.role "
+        "SELECT u.id, u.password_hash, u.is_active, t.id, m.role, "
+        "COALESCE(u.locked_until > now(), false) "
         "FROM {c}.users u "
         "LEFT JOIN {c}.tenants t ON t.slug = %s "
         "LEFT JOIN {c}.memberships m ON m.user_id = u.id AND m.tenant_id = t.id "
         "WHERE u.username = %s"), (tenant_slug, username))
     row = cur.fetchone()
     ok = verify_password(password, row[1] if row else _DUMMY_HASH)
-    if not (row and ok and row[2] and row[4]):
+    if row and not ok:
+        _record_failure(cur, row[0])
+    if not (row and ok and row[2] and row[4]) or row[5]:
         return None
+    cur.execute(_q("UPDATE {c}.users SET failed_logins = 0, first_failed_at = NULL, "
+                   "locked_until = NULL WHERE id = %s"), (row[0],))
     user_id, tenant_id, role = row[0], row[3], row[4]
     token = secrets.token_urlsafe(32)
     cur.execute(_q(
@@ -92,6 +98,21 @@ def login(cur, tenant_slug: str, username: str, password: str) -> dict | None:
             "tenant": tenant_slug, "role": role}
 
 
+def _record_failure(cur, user_id: int) -> None:
+    """Count a wrong password; lock the account once the window's limit is reached (#102)."""
+    cur.execute(_q(
+        "UPDATE {c}.users SET "
+        "failed_logins = CASE WHEN first_failed_at > now() - make_interval(mins => %(m)s) "
+        "                     THEN failed_logins + 1 ELSE 1 END, "
+        "first_failed_at = CASE WHEN first_failed_at > now() - make_interval(mins => %(m)s) "
+        "                       THEN first_failed_at ELSE now() END "
+        "WHERE id = %(id)s RETURNING failed_logins"),
+        {"m": settings.LOGIN_LOCKOUT_MINUTES, "id": user_id})
+    if cur.fetchone()[0] >= settings.LOGIN_MAX_FAILURES:
+        cur.execute(_q("UPDATE {c}.users SET locked_until = now() + make_interval(mins => %s) "
+                       "WHERE id = %s"), (settings.LOGIN_LOCKOUT_MINUTES, user_id))
+
+
 def session_for(cur, token: str) -> dict | None:
     """The live session for a token, with its user, tenant and role, else None."""
     cur.execute(_q(

docs/ENV_VARIABLES.md

@@ -41,6 +41,8 @@ Read by `api/config.py`. Set these in the shell before running
 | `CSV_UPLOADS_SCHEMA` | `csv_uploads` | No | Schema where per-CSV tables and the file registry live. |
 | `CONTROL_SCHEMA` | `control` | No | Schema for tenants, users, memberships and sign-in sessions (M2, #92). |
 | `SESSION_TTL_HOURS` | `12` | No | How long a sign-in session lasts before the user must sign in again. |
+| `LOGIN_MAX_FAILURES` | `10` | No | Failed sign-ins within the window that lock an account (#102). |
+| `LOGIN_LOCKOUT_MINUTES` | `15` | No | Length of the failure window and of the lock. |
 | `TE_SCHEMA` | `te_dev` | No | Schema holding the 12 fixed T&E tables. Switch to `te_test`, `te_staging`, or `te_prod` per environment. |
 | `MAX_UPLOAD_BYTES` | `52428800` (50 MB) | No | Maximum file size accepted by the upload endpoint. |
 

scripts/lint.sh

@@ -18,6 +18,7 @@ PY_SRC=(
     alembic/versions/0002_registry_unique.py
     alembic/versions/0003_import_counts.py
     alembic/versions/0004_control_schema.py
+    alembic/versions/0005_login_lockout.py
     api/auth.py
     api/services/accounts.py
     api/routers/auth_routes.py

tests/test_accounts.py

@@ -40,8 +40,9 @@ def test_malformed_hash_fails_closed(self):
             self.assertFalse(accounts.verify_password(_PW, bad), bad)
 
 
-@pytest.mark.integration
-class SignIn(unittest.TestCase):
+class _Accounts(unittest.TestCase):
+    """Shared setup: one tenant with one uploader, plus an unrelated tenant."""
+
     @classmethod
     def setUpClass(cls):
         cls.ctx = TestClient(app)
@@ -78,6 +79,9 @@ def _login(self, tenant=None, username=None, password=_PW):
     def _me(self, token: str):
         return self.client.get("/api/auth/me", headers={"Authorization": f"Bearer {token}"})
 
+
+@pytest.mark.integration
+class SignIn(_Accounts):
     def test_login_returns_a_session_for_the_tenant_and_role(self):
         res = self._login()
         self.assertEqual(res.status_code, 200, res.text)
@@ -133,3 +137,37 @@ def test_tokens_are_stored_only_as_hashes(self):
             stored = [r[0] for r in cur.fetchall()]
         self.assertEqual(len(stored), 1)
         self.assertNotIn(token, stored)
+
+
+@pytest.mark.integration
+class Lockout(_Accounts):
+    """Accounts lock after LOGIN_MAX_FAILURES wrong passwords in the window (#102)."""
+
+    def _fail(self, n: int) -> None:
+        for _ in range(n):
+            self.assertEqual(self._login(password="wrong").status_code, 401)
+
+    def test_eleventh_attempt_is_refused_even_with_the_right_password(self):
+        self._fail(settings.LOGIN_MAX_FAILURES)
+        res = self._login()
+        self.assertEqual((res.status_code, res.json()["detail"]),
+                         (401, "Invalid tenant, username or password"))
+
+    def test_one_short_of_the_limit_still_signs_in_and_resets_the_count(self):
+        self._fail(settings.LOGIN_MAX_FAILURES - 1)
+        self.assertEqual(self._login().status_code, 200)
+        self._fail(settings.LOGIN_MAX_FAILURES - 1)
+        self.assertEqual(self._login().status_code, 200, "a success must reset the count")
+
+    def test_sign_in_works_again_after_the_lock_expires(self):
+        self._fail(settings.LOGIN_MAX_FAILURES)
+        self._exec("UPDATE {c}.users SET locked_until = now() - interval '1 second' WHERE id = %s",
+                   self.user_id)
+        self.assertEqual(self._login().status_code, 200)
+
+    def test_failures_outside_the_window_start_a_new_count(self):
+        self._fail(settings.LOGIN_MAX_FAILURES - 1)
+        self._exec("UPDATE {c}.users SET first_failed_at = now() - make_interval(mins => %s) "
+                   "WHERE id = %s", settings.LOGIN_LOCKOUT_MINUTES + 1, self.user_id)
+        self._fail(1)
+        self.assertEqual(self._login().status_code, 200)

tests/test_migrations.py

@@ -148,7 +148,7 @@ def setUp(self):
     def test_bootstrap_brings_the_schema_to_head(self):
         with Conn() as conn, conn.cursor() as cur:
             cur.execute(f'SELECT version_num FROM "{settings.UPLOADS_SCHEMA}".alembic_version')
-            self.assertEqual(cur.fetchone()[0], "0004")
+            self.assertEqual(cur.fetchone()[0], "0005")
 
     def test_duplicate_hash_is_rejected_by_the_database(self):
         insert = (f'INSERT INTO "{settings.UPLOADS_SCHEMA}".csv_files (file_name, file_hash, table_name, mode) '

tests/test_tenant_isolation.py

@@ -150,7 +150,7 @@ def test_tenant_schema_is_at_head_with_its_own_registry(self):
         with Conn() as conn, conn.cursor() as cur:
             cur.execute(sql.SQL("SELECT version_num FROM {}.alembic_version").format(
                 sql.Identifier(schema_for(self.slugs["b"]))))
-            self.assertEqual(cur.fetchone()[0], "0004")
+            self.assertEqual(cur.fetchone()[0], "0005")
             cur.execute("SELECT table_name FROM information_schema.tables WHERE table_schema = %s",
                         (schema_for(self.slugs["b"]),))
             self.assertTrue({"csv_files", "audit_log"} <= {r[0] for r in cur.fetchall()})