Add login lockout and encrypt Player.totp_secret/auth_token at rest
Per-username lockout (5 failed TOTP attempts / 5 min) stops account-targeted brute force regardless of source IP. Player.totp_secret is now Fernet- encrypted (ENCRYPTION_KEY env, db/crypto.py) instead of stored in plaintext, and auth_token is stored as a SHA-256 hash rather than the raw session token. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
+84
-2
@@ -12,14 +12,21 @@ import unittest
|
||||
import uuid
|
||||
from types import SimpleNamespace
|
||||
|
||||
# Nastav DB PRED importom db/api modulov -- engine sa vytvara pri importe.
|
||||
# Nastav DB/ENCRYPTION_KEY PRED importom db/api modulov -- engine sa vytvara pri importe.
|
||||
_DB_FILE = os.path.join(tempfile.gettempdir(), f"bridzik_test_{uuid.uuid4().hex}.db")
|
||||
os.environ["DATABASE_URL"] = "sqlite+aiosqlite:///" + _DB_FILE.replace("\\", "/")
|
||||
|
||||
from cryptography.fernet import Fernet # noqa: E402
|
||||
|
||||
os.environ.setdefault("ENCRYPTION_KEY", Fernet.generate_key().decode())
|
||||
|
||||
import pyotp # noqa: E402
|
||||
from sqlalchemy import select # noqa: E402
|
||||
|
||||
from api import auth, history # noqa: E402
|
||||
from db.db import init_db # noqa: E402
|
||||
from db import crypto # noqa: E402
|
||||
from db.db import async_session, init_db # noqa: E402
|
||||
from db.models import Player # noqa: E402
|
||||
|
||||
|
||||
def run(coro):
|
||||
@@ -76,6 +83,32 @@ class HistoryCase(unittest.TestCase):
|
||||
with self.assertRaises(auth.AuthError):
|
||||
run(auth.login(username, "000000"))
|
||||
|
||||
def test_login_lockout_after_repeated_failures(self):
|
||||
username = "bob_" + uuid.uuid4().hex[:6]
|
||||
data = run(auth.register_account(username))
|
||||
|
||||
for _ in range(auth._LOGIN_ATTEMPT_LIMIT):
|
||||
with self.assertRaises(auth.AuthError):
|
||||
run(auth.login(username, "000000"))
|
||||
|
||||
# Lockout odmietne aj spravny kod, kym neubehne okno
|
||||
code = pyotp.TOTP(data["secret"]).now()
|
||||
with self.assertRaises(auth.AuthError):
|
||||
run(auth.login(username, code))
|
||||
|
||||
def test_successful_login_clears_failed_attempts(self):
|
||||
username = "carol_" + uuid.uuid4().hex[:6]
|
||||
data = run(auth.register_account(username))
|
||||
|
||||
for _ in range(auth._LOGIN_ATTEMPT_LIMIT - 1):
|
||||
with self.assertRaises(auth.AuthError):
|
||||
run(auth.login(username, "000000"))
|
||||
|
||||
code = pyotp.TOTP(data["secret"]).now()
|
||||
ident = run(auth.login(username, code))
|
||||
self.assertEqual(ident["username"], username)
|
||||
self.assertNotIn(username, auth._failed_attempts)
|
||||
|
||||
def test_record_rounds_and_idempotency(self):
|
||||
ids = self._make_players()
|
||||
gid = str(uuid.uuid4())
|
||||
@@ -191,6 +224,55 @@ class HistoryCase(unittest.TestCase):
|
||||
self.assertEqual(mine["my_points"], 12)
|
||||
self.assertEqual(len(mine["players"]), 4)
|
||||
|
||||
def test_totp_secret_stored_encrypted_not_plaintext(self):
|
||||
username = "dave_" + uuid.uuid4().hex[:6]
|
||||
data = run(auth.register_account(username))
|
||||
|
||||
async def _raw_secret():
|
||||
async with async_session() as session:
|
||||
return await session.scalar(
|
||||
select(Player.totp_secret).where(Player.username == username)
|
||||
)
|
||||
|
||||
stored = run(_raw_secret())
|
||||
self.assertNotEqual(stored, data["secret"])
|
||||
self.assertEqual(crypto.decrypt(stored), data["secret"])
|
||||
|
||||
def test_auth_token_stored_hashed_not_plaintext(self):
|
||||
username = "erin_" + uuid.uuid4().hex[:6]
|
||||
data = run(auth.register_account(username))
|
||||
ident = run(auth.login(username, pyotp.TOTP(data["secret"]).now()))
|
||||
|
||||
async def _raw_token():
|
||||
async with async_session() as session:
|
||||
return await session.scalar(
|
||||
select(Player.auth_token).where(Player.username == username)
|
||||
)
|
||||
|
||||
stored = run(_raw_token())
|
||||
self.assertNotEqual(stored, ident["token"])
|
||||
self.assertEqual(stored, crypto.hash_token(ident["token"]))
|
||||
|
||||
|
||||
class CryptoCase(unittest.TestCase):
|
||||
def test_encrypt_decrypt_roundtrip(self):
|
||||
secret = "JBSWY3DPEHPK3PXP"
|
||||
ciphertext = crypto.encrypt(secret)
|
||||
self.assertNotEqual(ciphertext, secret)
|
||||
self.assertEqual(crypto.decrypt(ciphertext), secret)
|
||||
|
||||
def test_hash_token_is_deterministic_and_distinct(self):
|
||||
self.assertEqual(crypto.hash_token("abc"), crypto.hash_token("abc"))
|
||||
self.assertNotEqual(crypto.hash_token("abc"), crypto.hash_token("abd"))
|
||||
|
||||
def test_missing_key_raises(self):
|
||||
saved = os.environ.pop("ENCRYPTION_KEY")
|
||||
try:
|
||||
with self.assertRaises(RuntimeError):
|
||||
crypto.encrypt("x")
|
||||
finally:
|
||||
os.environ["ENCRYPTION_KEY"] = saved
|
||||
|
||||
|
||||
def tearDownModule():
|
||||
from db.db import engine
|
||||
|
||||
Reference in New Issue
Block a user