ipam_control/tests/test_journal.py

import ipaddress
import uuid

import httpx

from app.request_context import parse_networks, resolve_client_ip
from app.security import hash_password
from tests.conftest import BASE


def test_journal_search_and_filters(client, org):
    cidr = f"fd00:{uuid.uuid4().int % 0xF000 + 0x1000:x}:{uuid.uuid4().int % 0xF000 + 0x1000:x}::/64"  # уникальный для запуска (журнал общий); группы без ведущих нулей — иначе PostgreSQL CIDR нормализует запись и поиск по тексту промахивается
    client.post("/prefixes", json={"organization_id": org["id"], "vrf_id": org["vrf_id"], "prefix": cidr})
    hit = client.get("/audit", params={"q": cidr}).json()
    assert hit["total"] >= 1
    entry = hit["items"][0]
    assert entry["event_type"] == "prefix.created" and entry["actor"] == "ui:admin" and "создан" in entry["message"]
    assert client.get("/audit", params={"q": "evt_" + entry["short_id"]}).json()["total"] >= 1  # поиск по ID
    assert client.get("/audit", params={"q": cidr, "event_type": "prefix.deleted"}).json()["total"] == 0
    assert client.get("/audit", params={"q": cidr, "actor": "system"}).json()["total"] == 0
    assert client.get("/audit", params={"q": cidr, "date_from": "2999-01-01"}).json()["total"] == 0
    assert client.get(f"/audit/{entry['uid']}").json()["id"] == entry["id"]
    assert client.get("/audit", params={"q": "100%_"}).json()["total"] == 0  # спецсимволы LIKE экранируются


def test_rotation_by_age_and_count(client, db):
    old = db.execute("INSERT INTO audit_log (ts, username, entity_type, entity_label, action, message) "
                     "VALUES (now() - interval '200 days', 'admin', 'prefix', 'rot-old', 'created', 'ROT-OLD') RETURNING uid").fetchone()[0]
    try:
        assert client.put("/journal/settings", json={"retention_days": 30, "max_entries": 100000}).json()["deleted"] >= 1
        assert client.get(f"/audit/{old}").status_code == 404  # старая запись ушла
        total = client.get("/audit/summary").json()["total"]
        r = client.put("/journal/settings", json={"retention_days": 30, "max_entries": total - 3}).json()
        assert r["deleted"] == 5  # 4 лишних (с учётом записи о смене настроек) + место под сводную запись
        assert client.get("/audit/summary").json()["total"] == total - 3
        assert client.get("/audit", params={"event_type": "journal.rotated"}).json()["total"] >= 1
    finally:
        client.put("/journal/settings", json={"retention_days": 90, "max_entries": 100000})


def test_clear_requires_password_and_locks_out(db):
    name, pw = f"tmp-{uuid.uuid4().hex[:6]}", "tmp-pass-123"
    db.execute("INSERT INTO users (username, password_hash, role, is_active) VALUES (%s, %s, 'superadmin', true)", (name, hash_password(pw)))  # изменение 033: очистка журнала — только superadmin
    try:
        c = httpx.Client(base_url=BASE)
        c.headers["Authorization"] = "Bearer " + c.post("/auth/login", json={"username": name, "password": pw}).json()["access_token"]
        first = c.post("/journal/clear", json={"password": "wrong"})
        assert first.status_code == 403 and first.json()["attempts_left"] == 4
        codes = [c.post("/journal/clear", json={"password": "wrong"}).status_code for _ in range(4)]
        assert codes == [403, 403, 403, 429]  # пятая неудача — блокировка
        locked = c.post("/journal/clear", json={"password": pw})  # даже верный пароль при блокировке отклоняется
        assert locked.status_code == 429 and locked.json()["retry_after_seconds"] > 0
    finally:
        db.execute("DELETE FROM users WHERE username = %s", (name,))


def test_client_ip_and_request_meta_recorded(client, org):
    hdr = {"User-Agent": "ipam-tests/1.0", "X-Forwarded-For": "203.0.113.9"}  # подделка XFF от недоверенного пира
    cidr = f"fd00:{uuid.uuid4().int % 0xF000 + 0x1000:x}:{uuid.uuid4().int % 0xF000 + 0x1000:x}::/64"
    client.post("/prefixes", headers=hdr, json={"organization_id": org["id"], "vrf_id": org["vrf_id"], "prefix": cidr})
    entry = client.get("/audit", params={"q": cidr}).json()["items"][0]
    assert ipaddress.ip_address(entry["client_ip"])
    assert entry["client_ip"] != "203.0.113.9"  # заголовок не доверенного прокси игнорируется
    assert entry["meta"]["user_agent"] == "ipam-tests/1.0" and entry["meta"]["method"] == "POST" and entry["meta"]["path"] == "/api/v1/prefixes"
    assert client.get("/audit", params={"q": cidr, "client_ip": entry["client_ip"]}).json()["total"] >= 1  # точный IP
    assert client.get("/audit", params={"q": cidr, "client_ip": "203.0.113.0/24"}).json()["total"] == 0
    assert client.get("/audit", params={"client_ip": "не-ip"}).status_code == 422
    rotated = client.get("/audit", params={"event_type": "journal.rotated", "limit": 1}).json()["items"]
    assert all(r["client_ip"] is None for r in rotated)  # системные события — без IP


def test_resolve_client_ip_trusts_forwarded_header_only_from_proxies():
    trusted = parse_networks("10.0.0.0/8, 192.168.1.5")
    assert resolve_client_ip("198.51.100.7", "203.0.113.9", trusted) == "198.51.100.7"  # недоверенный пир: XFF игнорируется
    assert resolve_client_ip("10.1.2.3", "203.0.113.9, 10.9.9.9", trusted) == "203.0.113.9"  # справа первый недоверенный
    assert resolve_client_ip("10.1.2.3", "1.1.1.1, 203.0.113.9, 10.9.9.9", trusted) == "203.0.113.9"  # клиент не может «подставить» левый адрес
    assert resolve_client_ip("10.1.2.3", "мусор", trusted) == "10.1.2.3"
    assert resolve_client_ip("::ffff:192.0.2.4", None, trusted) == "192.0.2.4"  # IPv4-mapped
    assert resolve_client_ip(None, None, trusted) is None