saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
119 lines
4.3 KiB
Python
119 lines
4.3 KiB
Python
"""Per-account known-IP ledger: pure data capture on every recorded sign-in
|
|
(real login and the local-dev bypass), one row per distinct (user, ip) pair,
|
|
touched rather than duplicated on repeat visits from the same IP. Nothing
|
|
currently reads this data - it's the foundation a later "new IP" security
|
|
feature would query against."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.db.base import ensure_aware_utc
|
|
from app.main import app
|
|
from app.models.user import LOCAL_DEV_USER_ID
|
|
from app.repositories.user_known_ip_repository import UserKnownIpRepository
|
|
from app.repositories.user_repository import UserRepository
|
|
|
|
|
|
def _unique_email() -> str:
|
|
return f"user-{uuid.uuid4().hex[:12]}@example.com"
|
|
|
|
|
|
def _unique_ip() -> str:
|
|
return f"10.{uuid.uuid4().int % 250}.{uuid.uuid4().int % 250}.{uuid.uuid4().int % 250}"
|
|
|
|
|
|
# --- Repository --------------------------------------------------------
|
|
|
|
|
|
async def test_record_login_creates_a_row_for_a_new_ip_and_reports_it_as_new(db_session):
|
|
repo = UserKnownIpRepository(db_session)
|
|
user_id = uuid.uuid4()
|
|
now = datetime.now(UTC)
|
|
|
|
is_new = await repo.record_login(user_id, "203.0.113.5", now)
|
|
assert is_new is True
|
|
|
|
row = await repo.get(user_id, "203.0.113.5")
|
|
assert row is not None
|
|
assert ensure_aware_utc(row.first_seen_at) == now
|
|
assert ensure_aware_utc(row.last_seen_at) == now
|
|
|
|
|
|
async def test_record_login_touches_last_seen_instead_of_duplicating(db_session):
|
|
repo = UserKnownIpRepository(db_session)
|
|
user_id = uuid.uuid4()
|
|
first_seen = datetime.now(UTC) - timedelta(days=1)
|
|
second_visit = datetime.now(UTC)
|
|
|
|
await repo.record_login(user_id, "203.0.113.6", first_seen)
|
|
is_new = await repo.record_login(user_id, "203.0.113.6", second_visit)
|
|
|
|
assert is_new is False
|
|
rows = await repo.list_for_user(user_id)
|
|
assert len(rows) == 1
|
|
assert ensure_aware_utc(rows[0].first_seen_at) == first_seen
|
|
assert ensure_aware_utc(rows[0].last_seen_at) == second_visit
|
|
|
|
|
|
async def test_a_second_distinct_ip_creates_a_second_row(db_session):
|
|
repo = UserKnownIpRepository(db_session)
|
|
user_id = uuid.uuid4()
|
|
now = datetime.now(UTC)
|
|
|
|
await repo.record_login(user_id, "203.0.113.7", now)
|
|
await repo.record_login(user_id, "203.0.113.8", now)
|
|
|
|
rows = await repo.list_for_user(user_id)
|
|
assert {r.ip_address for r in rows} == {"203.0.113.7", "203.0.113.8"}
|
|
|
|
|
|
# --- Wired into real sign-in flows --------------------------------------
|
|
|
|
|
|
async def test_real_login_records_the_client_ip(db_session):
|
|
ip = _unique_ip()
|
|
email = _unique_email()
|
|
with TestClient(app, client=(ip, 51234)) as c:
|
|
c.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": "correct-horse-1", "display_name": "T"},
|
|
)
|
|
login_resp = c.post(
|
|
"/api/v1/auth/login", json={"email": email, "password": "correct-horse-1"}
|
|
)
|
|
assert login_resp.status_code == 200
|
|
|
|
user = await UserRepository(db_session).get_by_email(email)
|
|
rows = await UserKnownIpRepository(db_session).list_for_user(user.id)
|
|
assert [r.ip_address for r in rows] == [ip]
|
|
|
|
|
|
async def test_repeat_login_from_the_same_ip_does_not_duplicate_the_row(db_session):
|
|
ip = _unique_ip()
|
|
email = _unique_email()
|
|
with TestClient(app, client=(ip, 51235)) as c:
|
|
c.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": "correct-horse-1", "display_name": "T"},
|
|
)
|
|
c.post("/api/v1/auth/login", json={"email": email, "password": "correct-horse-1"})
|
|
c.post("/api/v1/auth/login", json={"email": email, "password": "correct-horse-1"})
|
|
|
|
user = await UserRepository(db_session).get_by_email(email)
|
|
rows = await UserKnownIpRepository(db_session).list_for_user(user.id)
|
|
assert len(rows) == 1
|
|
|
|
|
|
async def test_local_dev_sign_in_records_the_known_ip(local_mode_client, db_session):
|
|
# local_mode_client's fixed loopback peer (see conftest.py) should show
|
|
# up as a known IP for the local-dev account after this request.
|
|
resp = local_mode_client.get("/api/v1/auth/me")
|
|
assert resp.status_code == 200
|
|
|
|
rows = await UserKnownIpRepository(db_session).list_for_user(LOCAL_DEV_USER_ID)
|
|
assert "127.0.0.1" in {r.ip_address for r in rows}
|