ayurishchevandClaude Opus 5.5
8384c2c311
034 Публикация стенда как https://rxipam.rxmsk.ru через общий Caddy хоста (Caddyfile вне репозитория): UI и /api/v1 доступны из интернета, /docs, /redoc, /openapi.json — только из частных сетей (RFC 1918/4193, loopback). Caddy проксирует на опубликованный порт приложения — общая Docker-сеть отклонена из-за коллизии имён app/db с Nextcloud. TRUSTED_PROXIES=172.16.0.0/12: реальный IP клиента из X-Forwarded-For. Интеграционный тест журнала больше не проверяет подделку XFF с хоста (хост за Caddy доверенный), логику покрывает unit-тест. 035 Дата и IP последнего входа в разделе «Пользователи»: users.last_login_at / last_login_ip (миграция 0014 с заполнением из журнала), запись при успешном входе, UserOut, колонка в UI. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
80 lines
5.9 KiB
Python
80 lines
5.9 KiB
Python
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):
|
|
# доверие к X-Forwarded-For зависит от TRUSTED_PROXIES стенда (за Caddy хост доверенный, изменение 034) — логику проверяет unit-тест ниже
|
|
hdr = {"User-Agent": "ipam-tests/1.0"}
|
|
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["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
|