- 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>
115 lines
3.6 KiB
Python
115 lines
3.6 KiB
Python
"""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"]()
|