285 lines
13 KiB
Python
285 lines
13 KiB
Python
import asyncio
|
|
import os
|
|
import pty
|
|
import secrets
|
|
import select as io_select
|
|
import signal
|
|
import sys
|
|
import time
|
|
from datetime import timedelta
|
|
from uuid import uuid4
|
|
|
|
import httpx
|
|
import pytest
|
|
from sqlalchemy import event, func, select, update
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from kf_oj.api.app import create_app
|
|
from kf_oj.contracts.identity import LoginRequest
|
|
from kf_oj.core.settings import Settings
|
|
from kf_oj.modules.identity.models import Account, Session
|
|
from kf_oj.modules.identity.passwords import hash_password, verify_password
|
|
from kf_oj.modules.identity.public import AuthenticationFailed, LoginThrottled
|
|
from kf_oj.modules.identity.service import IdentityService, token_digest
|
|
|
|
|
|
def restrict_platform_role(engine):
|
|
@event.listens_for(engine.sync_engine, "connect")
|
|
def apply_role(connection, _):
|
|
cursor = connection.cursor()
|
|
cursor.execute("SET ROLE kf_oj_platform")
|
|
cursor.close()
|
|
|
|
|
|
def test_password_policy_and_credentials_redacted():
|
|
password = secrets.token_hex(16)
|
|
hashed = hash_password("fixture_user", password)
|
|
assert hashed.startswith("$argon2id$v=19$m=19456,t=2,p=1$")
|
|
assert verify_password(hashed, password)
|
|
assert not verify_password(hashed, secrets.token_hex(16))
|
|
assert not verify_password("$argon2id$v=19$m=999999999,t=2,p=1$invalid", password)
|
|
assert password not in repr(LoginRequest(username="fixture_user", password=password))
|
|
with pytest.raises(ValueError):
|
|
hash_password("not canonical", password)
|
|
with pytest.raises(ValueError):
|
|
hash_password("fixture_user", "too short")
|
|
|
|
|
|
def test_production_cookie_configuration_requires_explicit_https_origins():
|
|
with pytest.raises(ValueError):
|
|
Settings(_env_file=None, environment="production")
|
|
for invalid in ("https://oj.example/path", "https://user@oj.example", "*", "null"):
|
|
with pytest.raises(ValueError):
|
|
Settings(_env_file=None, browser_origins=(invalid,))
|
|
|
|
|
|
def test_real_login_session_revocation_expiry_and_throttle(postgres_url):
|
|
async def run():
|
|
engine = create_async_engine(postgres_url)
|
|
restrict_platform_role(engine)
|
|
sessions = async_sessionmaker(engine, expire_on_commit=False)
|
|
identity = IdentityService(sessions, lifetime_seconds=300)
|
|
password = secrets.token_hex(16)
|
|
try:
|
|
principal = await identity.create_account("fixture_user", password)
|
|
with pytest.raises(AuthenticationFailed):
|
|
await identity.login("fixture_user", secrets.token_hex(16), peer="fixture-peer")
|
|
with pytest.raises(AuthenticationFailed):
|
|
await identity.login("absent_user", password, peer="fixture-peer")
|
|
issued = await identity.login("fixture_user", password, peer="fixture-peer")
|
|
assert await identity.authenticate(issued.token) == principal
|
|
assert issued.token not in repr(issued) and issued.csrf_token not in repr(issued)
|
|
with pytest.raises(AuthenticationFailed):
|
|
await identity.authenticate(str(principal.account_id))
|
|
with pytest.raises(AuthenticationFailed):
|
|
await identity.authenticate(issued.token, csrf_token="f" * 64)
|
|
await identity.logout(issued.token, issued.csrf_token)
|
|
with pytest.raises(AuthenticationFailed):
|
|
await identity.authenticate(issued.token)
|
|
issued = await identity.login("fixture_user", password, peer="expiry-peer")
|
|
async with sessions.begin() as session:
|
|
# Keep the expiry constraint valid while moving the entire session into the past.
|
|
await session.execute(
|
|
update(Session).values(
|
|
created_at=func.now() - timedelta(minutes=10),
|
|
expires_at=func.now() - timedelta(minutes=5),
|
|
)
|
|
)
|
|
with pytest.raises(AuthenticationFailed):
|
|
await identity.authenticate(issued.token)
|
|
issued = await identity.login("fixture_user", password, peer="disable-peer")
|
|
assert await identity.disable_account("fixture_user")
|
|
with pytest.raises(AuthenticationFailed):
|
|
await identity.authenticate(issued.token)
|
|
with pytest.raises(AuthenticationFailed):
|
|
await identity.login("fixture_user", password, peer="disable-peer")
|
|
assert not await identity.disable_account("unknown_user")
|
|
# Exactly ten real failed attempts consume the same persistent peer budget.
|
|
for _ in range(10):
|
|
with pytest.raises(AuthenticationFailed):
|
|
await identity.login("absent_user", password, peer="limited-peer")
|
|
with pytest.raises(LoginThrottled):
|
|
await identity.login("absent_user", password, peer="limited-peer")
|
|
async with sessions() as session:
|
|
account = await session.scalar(select(Account))
|
|
assert password != account.password_hash and password not in account.password_hash
|
|
rows = list(await session.scalars(select(Session)))
|
|
assert all(row.token_hash != issued.token for row in rows)
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_real_http_cookie_csrf_and_identity_spoofing_rejection(postgres_url, caplog):
|
|
async def run():
|
|
app = create_app(
|
|
Settings(
|
|
_env_file=None,
|
|
environment="production",
|
|
database_url=postgres_url,
|
|
browser_origins=("https://test",),
|
|
)
|
|
)
|
|
async with app.router.lifespan_context(app):
|
|
restrict_platform_role(app.state.sessions.kw["bind"])
|
|
password = secrets.token_hex(16)
|
|
first = await app.state.identity.create_account("first_user", password)
|
|
second = await app.state.identity.create_account("second_user", secrets.token_hex(16))
|
|
headers = {"Origin": "https://test"}
|
|
async with httpx.AsyncClient(
|
|
transport=httpx.ASGITransport(app=app), base_url="https://test"
|
|
) as client:
|
|
body = {"username": "first_user", "password": password}
|
|
assert (await client.get("/api/v1/auth/me")).status_code == 401
|
|
assert (await client.post("/api/v1/auth/login", json=body)).status_code == 403
|
|
assert (
|
|
await client.post(
|
|
"/api/v1/auth/login",
|
|
json=body,
|
|
headers={"Origin": "https://hostile.example"},
|
|
)
|
|
).status_code == 403
|
|
invalid = await client.post(
|
|
"/api/v1/auth/login",
|
|
headers=headers,
|
|
json=body | {"user_id": str(second.account_id), "role": "admin"},
|
|
)
|
|
assert invalid.status_code == 422
|
|
assert password not in invalid.text and "admin" not in invalid.text
|
|
response = await client.post("/api/v1/auth/login", json=body, headers=headers)
|
|
assert response.status_code == 200
|
|
assert response.json()["user"]["id"] == str(first.account_id)
|
|
assert (
|
|
await client.get(
|
|
"/api/v1/auth/me", headers={"X-User-ID": str(second.account_id)}
|
|
)
|
|
).json()["id"] == str(first.account_id)
|
|
cookie = response.headers.get_list("set-cookie")[0]
|
|
assert "HttpOnly" in cookie and "Secure" in cookie and "SameSite=strict" in cookie
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert client.cookies.get("kf_csrf") == response.json()["csrf_token"]
|
|
assert (
|
|
await client.post("/api/v1/auth/logout", headers=headers)
|
|
).status_code == 403
|
|
assert (
|
|
await client.post(
|
|
"/api/v1/auth/logout", headers=headers | {"X-CSRF-Token": "f" * 64}
|
|
)
|
|
).status_code == 403
|
|
assert (
|
|
await client.post(
|
|
"/api/v1/auth/logout",
|
|
headers=headers | {"X-CSRF-Token": response.json()["csrf_token"]},
|
|
)
|
|
).status_code == 204
|
|
assert (await client.get("/api/v1/auth/me")).status_code == 401
|
|
assert not client.cookies.get("kf_session") and not client.cookies.get("kf_csrf")
|
|
assert (
|
|
password not in caplog.text and response.json()["csrf_token"] not in caplog.text
|
|
)
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_real_concurrent_logins_keep_bounded_active_sessions(postgres_url):
|
|
async def run():
|
|
engine = create_async_engine(postgres_url)
|
|
restrict_platform_role(engine)
|
|
sessions = async_sessionmaker(engine, expire_on_commit=False)
|
|
identity = IdentityService(sessions)
|
|
password = secrets.token_hex(16)
|
|
try:
|
|
await identity.create_account("fixture_user", password)
|
|
# Independent peer budgets exercise session bounds without exhausting login retries.
|
|
issued = []
|
|
for _ in range(12):
|
|
issued.append(await identity.login("fixture_user", password, peer=uuid4().hex))
|
|
with pytest.raises(AuthenticationFailed):
|
|
await identity.authenticate(issued[0].token)
|
|
results = await asyncio.gather(
|
|
*(identity.login("fixture_user", password, peer=uuid4().hex) for _ in range(3)),
|
|
return_exceptions=True,
|
|
)
|
|
assert sum(isinstance(result, LoginThrottled) for result in results) == 1
|
|
assert sum(not isinstance(result, Exception) for result in results) == 2
|
|
async with sessions() as session:
|
|
assert await session.scalar(select(func.count()).select_from(Session)) == 10
|
|
assert (
|
|
await session.scalar(
|
|
select(func.count())
|
|
.select_from(Session)
|
|
.where(Session.token_hash == token_digest("0" * 64))
|
|
)
|
|
== 0
|
|
)
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_account_cli_hidden_password_and_disable(postgres_url):
|
|
password = secrets.token_hex(16)
|
|
# The CLI prompts through a real private PTY; the password is never an argument/environment.
|
|
pid, descriptor = pty.fork()
|
|
if pid == 0:
|
|
os.environ["KF_OJ_DATABASE_URL"] = postgres_url
|
|
os.execv(
|
|
sys.executable,
|
|
[sys.executable, "-m", "kf_oj.modules.identity.cli", "create", "cli_user"],
|
|
)
|
|
output = b""
|
|
deadline, prompts, status = time.monotonic() + 15, 0, None
|
|
try:
|
|
while time.monotonic() < deadline:
|
|
if io_select.select([descriptor], [], [], 0.1)[0]:
|
|
try:
|
|
block = os.read(descriptor, 4096)
|
|
except OSError:
|
|
block = b""
|
|
output += block
|
|
if prompts == 0 and b"Password (12" in output:
|
|
os.write(descriptor, password.encode() + b"\n")
|
|
prompts = 1
|
|
if prompts == 1 and b"Repeat password:" in output:
|
|
os.write(descriptor, password.encode() + b"\n")
|
|
prompts = 2
|
|
result, status = os.waitpid(pid, os.WNOHANG)
|
|
if result:
|
|
break
|
|
else:
|
|
os.kill(pid, signal.SIGTERM)
|
|
os.waitpid(pid, 0)
|
|
pytest.fail("account CLI deadline exceeded")
|
|
assert os.waitstatus_to_exitcode(status) == 0
|
|
assert prompts == 2 and b"Account created." in output
|
|
assert password.encode() not in output
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
async def inspect():
|
|
engine = create_async_engine(postgres_url)
|
|
sessions = async_sessionmaker(engine, expire_on_commit=False)
|
|
try:
|
|
service = IdentityService(sessions)
|
|
issued = await service.login("cli_user", password, peer="cli-fixture")
|
|
from kf_oj.modules.identity.cli import run
|
|
|
|
await run("disable", "cli_user")
|
|
with pytest.raises(AuthenticationFailed):
|
|
await service.authenticate(issued.token)
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
# CLI run uses the test DB only for this call; don't reuse a configured business database.
|
|
original = os.environ.get("KF_OJ_DATABASE_URL")
|
|
os.environ["KF_OJ_DATABASE_URL"] = postgres_url
|
|
try:
|
|
asyncio.run(inspect())
|
|
finally:
|
|
if original is None:
|
|
os.environ.pop("KF_OJ_DATABASE_URL", None)
|
|
else:
|
|
os.environ["KF_OJ_DATABASE_URL"] = original
|