fix(503-4 로그인 잠금): 5 번 틀림 뒤 잠금 · 경고 메일 — 잠금 판정 별도 UPDATE · 신뢰 기기 확인 읽기만

- record_failed_login: 한 UPDATE 안 왼→오 대입으로 4 번째에 잠기던 흠 — 잠금은 failures >= 5 별도 UPDATE
- has_trusted_device: last_used_at UPDATE 뺌 — 마무리 트랜잭션 _trust_device 가 갱신(커밋 1)
- 시험: 4 번 안 잠김 · 5 번 잠김 · 경고 메일 4/5 · 쿠키 재로그인 커밋 1 · 살아 있는 서버 손 시험 helper_503_4

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

common_util/common_util_auth_repository.py

@@ -212,13 +212,17 @@ async def record_failed_login(user_id: int) -> int:
     async with pool.acquire() as connection, connection.cursor() as cursor:
         await cursor.execute(
             """UPDATE users SET login_failures = login_failures + 1,
-                      last_failed_at = CURRENT_TIMESTAMP,
-                      account_locked_until = CASE WHEN login_failures + 1 >= 5
-                        THEN DATE_ADD(CURRENT_TIMESTAMP, INTERVAL 15 MINUTE)
-                        ELSE account_locked_until END
+                      last_failed_at = CURRENT_TIMESTAMP
                WHERE id = %s""",
             (user_id,),
         )
+        # 잠금은 따로 — 한 UPDATE 안에서는 MariaDB 가 앞 대입(+1)을 본 값으로 판정해 4 번째에 잠김
+        await cursor.execute(
+            """UPDATE users
+               SET account_locked_until = DATE_ADD(CURRENT_TIMESTAMP, INTERVAL 15 MINUTE)
+               WHERE id = %s AND login_failures >= 5""",
+            (user_id,),
+        )
         await cursor.execute("SELECT login_failures FROM users WHERE id = %s", (user_id,))
         failures = (await cursor.fetchone())[0]
         await connection.commit()
@@ -265,6 +269,7 @@ async def finish_login(
 
 
 async def has_trusted_device(user_id: int, token_hash: str) -> bool:
+    """읽기만 — last_used_at 갱신은 finish_login 트랜잭션의 _trust_device 가 함."""
     valid_after = datetime.utcnow() - timedelta(days=EMAIL_REVERIFY_DAYS)
     pool = get_db_pool()
     async with pool.acquire() as connection, connection.cursor() as cursor:
@@ -273,15 +278,7 @@ async def has_trusted_device(user_id: int, token_hash: str) -> bool:
                WHERE user_id = %s AND token_hash = %s AND last_used_at >= %s LIMIT 1""",
             (user_id, token_hash, valid_after),
         )
-        row = await cursor.fetchone()
-        if not row:
-            return False
-        await cursor.execute(
-            "UPDATE trusted_devices SET last_used_at = CURRENT_TIMESTAMP WHERE id = %s",
-            (row[0],),
-        )
-        await connection.commit()
-        return True
+        return bool(await cursor.fetchone())
 
 
 async def trust_device(

resources/tester/helper_503_4_lock_live.py

@@ -0,0 +1,114 @@
+"""503-4 로그인 잠금 횟수 · 쿠키 재로그인 — 살아 있는 서버(8001)에서 손으로 돌림.
+
+    ./venv/Scripts/python.exe resources/tester/helper_503_4_lock_live.py [time|lock]
+
+time = 쿠키 있는 재로그인 다섯 번 시간(중앙값) · last_used_at 갱신 확인
+lock = 일반 계정 4 번 틀림(안 잠김) → 5 번(잠김) → 6 번(423) · 끝나면 users 행 되돌림
+계정 = tmp/verify_accounts.json 「일반」(값은 출력 안 함).
+"""
+
+import asyncio
+import json
+import statistics
+import sys
+import time
+from pathlib import Path
+
+import aiomysql
+import httpx
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+
+from config.config_system import DB_HOST, DB_NAME, DB_PASSWORD, DB_PORT, DB_USER  # noqa: E402
+
+BASE = "http://localhost:8001/api"
+ACC = json.loads((ROOT / "tmp/verify_accounts.json").read_text(encoding="utf-8"))["일반"]
+COLS = "login_failures, last_failed_at, account_locked_until"
+
+
+async def _q(sql, args=()):
+    conn = await aiomysql.connect(
+        host=DB_HOST,
+        port=DB_PORT,
+        db=DB_NAME,
+        user=DB_USER,
+        password=DB_PASSWORD,
+        autocommit=True,
+        init_command="SET time_zone='+00:00'",
+    )
+    try:
+        async with conn.cursor() as cur:
+            await cur.execute(sql, args)
+            return await cur.fetchall()
+    finally:
+        conn.close()
+
+
+def q(sql, args=()):
+    return asyncio.run(_q(sql, args))
+
+
+def row():
+    return q(f"SELECT id, {COLS} FROM users WHERE email = %s", (ACC["email"],))[0]
+
+
+def login(client, password):
+    t = time.perf_counter()
+    r = client.post("/auth/login/request", json={"email": ACC["email"], "password": password})
+    return r.status_code, time.perf_counter() - t
+
+
+def run_time():
+    client = httpx.Client(base_url=BASE, timeout=60)
+    assert login(client, ACC["password"])[0] == 200
+    uid = row()[0]
+    times = []
+    for _ in range(5):
+        before = q("SELECT MAX(last_used_at) FROM trusted_devices WHERE user_id = %s", (uid,))
+        time.sleep(1.1)
+        code, dt = login(client, ACC["password"])
+        after = q("SELECT MAX(last_used_at) FROM trusted_devices WHERE user_id = %s", (uid,))
+        assert code == 200 and after[0][0] > before[0][0], (code, before, after)
+        times.append(dt)
+    print(
+        "cookie relogin s:",
+        [round(t, 3) for t in times],
+        "median",
+        round(statistics.median(times), 3),
+    )
+
+
+def run_lock():
+    orig = row()
+    uid = orig[0]
+    print("orig failures", orig[1], "locked", orig[3] is not None)
+    try:
+        q(
+            "UPDATE users SET login_failures = 0, last_failed_at = NULL,"
+            " account_locked_until = NULL WHERE id = %s",
+            (uid,),
+        )
+        client = httpx.Client(base_url=BASE, timeout=60)
+        for n in range(1, 7):
+            code, _ = login(client, "wrong-password-503-4")
+            r = row()
+            print(f"try {n}: http {code} failures {r[1]} locked {r[3] is not None}")
+            if n <= 4:
+                assert code == 401 and r[1] == n and r[3] is None
+            elif n == 5:
+                assert code == 401 and r[1] == 5 and r[3] is not None
+            else:
+                assert code == 423
+        print("LOCK OK")
+    finally:
+        q(
+            "UPDATE users SET login_failures = %s, last_failed_at = %s,"
+            " account_locked_until = %s WHERE id = %s",
+            (orig[1], orig[2], orig[3], uid),
+        )
+        print("restored", row()[1:] == orig[1:])
+
+
+if __name__ == "__main__":
+    {"time": run_time, "lock": run_lock}[sys.argv[1] if len(sys.argv) > 1 else "lock"]()

resources/tester/test_309_1_login.py

@@ -4,6 +4,7 @@
 - 마무리 쓰기 = 한 커넥션 · 한 트랜잭션(begin 1 · commit 1) · 실패 = rollback · 예외
 - bcrypt 확인 = 이벤트 루프 밖 스레드
 - 보안 그대로 — 5 번 틀림 15 분 잠금 · 잠긴 계정 423 · 실패 기록
+- 503-4 — 4 번 틀림 = 안 잠김 · 5 번 = 잠김 · 경고 메일 · 쿠키 재로그인 커밋 1
 """
 
 import asyncio
@@ -16,6 +17,7 @@ from fastapi import HTTPException
 
 from A06_Login import A06_Login_Router as router
 from common_util import common_util_auth_repository as repo
+from common_util.common_util_auth import DEVICE_TOKEN_COOKIE_NAME
 
 ROOT = Path(__file__).resolve().parents[2]
 
@@ -161,7 +163,7 @@ def test_bcrypt_스레드(monkeypatch):
 
 def test_잠금_그대로(monkeypatch):
     src = (ROOT / "common_util/common_util_auth_repository.py").read_text(encoding="utf-8")
-    assert "login_failures + 1 >= 5" in src and "INTERVAL 15 MINUTE" in src
+    assert "login_failures >= 5" in src and "INTERVAL 15 MINUTE" in src
     logged = []
 
     async def rec(*a):
@@ -177,3 +179,117 @@ def test_잠금_그대로(monkeypatch):
     with pytest.raises(HTTPException) as e:
         asyncio.run(router.request_login(_payload(), _Req()))
     assert e.value.status_code == 423 and logged == ["ACCOUNT_LOCKED"]
+
+
+class _UsersRow:
+    """users 한 행 흉내 — record_failed_login 의 SQL 을 실행 순서대로 반영."""
+
+    def __init__(self):
+        self.failures, self.locked = 0, False
+
+
+class _RowCursor(FakeCursor):
+    def __init__(self, conn, row):
+        super().__init__(conn)
+        self.row = row
+
+    async def execute(self, sql, args=None):
+        await super().execute(sql, args)
+        flat = " ".join(sql.split())
+        if "login_failures = login_failures + 1" in flat:
+            self.row.failures += 1
+        if "account_locked_until = DATE_ADD" in flat:
+            assert "login_failures >= 5" in flat and "+ 1" not in flat
+            if self.row.failures >= 5:
+                self.row.locked = True
+
+    async def fetchone(self):
+        return (self.row.failures,)
+
+
+def _fail_n(monkeypatch, n):
+    row, pool = _UsersRow(), FakePool()
+    pool.acquire = lambda: _with_cursor(pool, row)
+    monkeypatch.setattr(repo, "get_db_pool", lambda: pool)
+    for _ in range(n):
+        asyncio.run(repo.record_failed_login(7))
+    return row
+
+
+def _with_cursor(pool, row):
+    conn = FakeConn()
+    conn.cursor = lambda *a: _RowCursor(conn, row)
+    pool.conns.append(conn)
+    return conn
+
+
+def test_잠금_4번_안잠김(monkeypatch):
+    row = _fail_n(monkeypatch, 4)
+    assert row.failures == 4 and row.locked is False
+
+
+def test_잠금_5번_잠김(monkeypatch):
+    row = _fail_n(monkeypatch, 5)
+    assert row.failures == 5 and row.locked is True
+
+
+@pytest.mark.parametrize("failures,sent", [(4, 0), (5, 1)])
+def test_경고_메일(monkeypatch, failures, sent):
+    mails = []
+
+    async def failed(user_id):
+        return failures
+
+    async def noop(*a, **k):
+        return None
+
+    async def get_user(email):
+        return _user()
+
+    monkeypatch.setattr(router, "verify_password", lambda *a: False)
+    monkeypatch.setattr(router, "get_user_by_email", get_user)
+    monkeypatch.setattr(router, "record_failed_login", failed)
+    monkeypatch.setattr(router, "record_login", noop)
+    monkeypatch.setattr(router, "ADMIN_EMAIL", "admin@example.com")
+    monkeypatch.setattr(router, "send_email_background", lambda *a: mails.append(a))
+    with pytest.raises(HTTPException) as e:
+        asyncio.run(router.request_login(_payload(), _Req()))
+    assert e.value.status_code == 401 and len(mails) == sent
+    if sent:
+        assert mails[0][0] == "admin@example.com"
+
+
+class _TrustedCursor(FakeCursor):
+    async def fetchone(self):
+        return (11,)
+
+
+def test_쿠키_재로그인_커밋_1(monkeypatch):
+    from datetime import datetime
+
+    pool = FakePool()
+    orig = pool.acquire
+
+    def acquire():
+        conn = orig()
+        conn.cursor = lambda *a: _TrustedCursor(conn)
+        return conn
+
+    pool.acquire = acquire
+
+    async def get_user(email):
+        return _user(last_email_verified_at=datetime.utcnow())
+
+    class Req(_Req):
+        cookies = {DEVICE_TOKEN_COOKIE_NAME: "tok"}
+
+    monkeypatch.setattr(repo, "get_db_pool", lambda: pool)
+    monkeypatch.setattr(router, "verify_password", lambda *a: True)
+    monkeypatch.setattr(router, "get_user_by_email", get_user)
+    monkeypatch.setattr(router, "AUTH_OTP_DISABLED", False)
+    res = asyncio.run(router.request_login(_payload(), Req()))
+    assert res.status_code == 200
+    kinds = [e[0] for c in pool.conns for e in c.log]
+    sqls = [e[1] for c in pool.conns for e in c.log if e[0] == "execute"]
+    assert kinds.count("commit") == 1 and kinds.count("begin") == 1
+    assert sum(s.startswith("UPDATE trusted_devices") for s in sqls) == 1