import json import pytest from cryptography.fernet import Fernet from fastapi.testclient import TestClient from app import process_lock, security from app.config import Settings, get_settings, insecure_settings from app.main import create_app from app.services import events from tests.conftest import ADMIN_PASSWORD, API_TOKEN from tests.helpers import _request def test_password_encryption_roundtrip(): token = security.encrypt("s3cret") assert token != "s3cret" and security.decrypt(token) == "s3cret" def test_api_auth_and_no_password_leak(): with TestClient(create_app()) as client: assert client.get("/api/v1/devices").status_code == 401 h = {"Authorization": f"Bearer {API_TOKEN}"} r = client.post("/api/v1/devices", headers=h, json={ "name": "r1", "host": "10.0.0.1", "username": "admin", "password": "pw"}) assert r.status_code == 201 and "password" not in r.text assert client.get("/api/v1/devices", headers=h).json()[0]["name"] == "r1" def test_insecure_settings_rejects_weak_secrets_without_leaking_them(): """Небезопасные секреты выявляются по каждому правилу; значения секретов в описание проблем не попадают.""" secret = "s3cr3t-value-must-not-leak-anywhere" bad = Settings(api_token="change-me", session_secret=secret[:10], admin_password=secret[:8], secret_key=secret) problems = insecure_settings(bad) assert len(problems) == 4 # все четыре секрета нарушают правила text = " ".join(problems) assert secret not in text and secret[:10] not in text and secret[:8] not in text ok = Settings(api_token="x" * 32, session_secret="y" * 32, admin_password="z" * 12, secret_key=Fernet.generate_key().decode()) assert insecure_settings(ok) == [] def test_insecure_settings_rejects_invalid_trusted_proxies(): """TRUSTED_PROXIES с записью не в формате CIDR — тоже отказ старта; значение (не секрет) видно в тексте проблемы.""" ok = Settings(api_token="x" * 32, session_secret="y" * 32, admin_password="z" * 12, secret_key=Fernet.generate_key().decode(), trusted_proxies="10.0.0.0/8, not-a-network") problems = insecure_settings(ok) assert len(problems) == 1 and "TRUSTED_PROXIES" in problems[0] and "not-a-network" in problems[0] def test_client_ip_trusted_proxies(monkeypatch): """X-Forwarded-For учитывается, только если TRUSTED_PROXIES не пуст и peer — из доверенной сети (иначе подделка заголовка бессмысленна); цепочка разбирается справа налево до первого недоверенного адреса; невалидная запись обрывает разбор на последнем валидном; работает и с IPv6.""" monkeypatch.setenv("TRUSTED_PROXIES", "") get_settings.cache_clear() assert security.client_ip(_request("1.2.3.4", "9.9.9.9")) == "1.2.3.4" # пусто — XFF полностью игнорируется monkeypatch.setenv("TRUSTED_PROXIES", "10.0.0.0/8") get_settings.cache_clear() assert security.client_ip(_request("9.9.9.9", "9.9.9.9")) == "9.9.9.9" # peer не доверенный — XFF игнорируется assert security.client_ip(_request("10.0.0.1", "1.2.3.4, 10.0.0.2")) == "1.2.3.4" # доверенный прокси пропущен assert security.client_ip(_request("10.0.0.1", "10.0.0.3, 10.0.0.2")) == "10.0.0.1" # все доверенные — peer assert security.client_ip(_request("10.0.0.1", "garbage, 10.0.0.2")) == "10.0.0.2" # мусор — последний валидный assert security.client_ip(_request("10.0.0.1", "garbage")) == "10.0.0.1" # мусор без валидных перед ним — peer assert security.client_ip(_request("10.0.0.1")) == "10.0.0.1" # заголовка нет assert security.client_ip(_request("10.0.0.1", "2001:db8::1, 10.0.0.2")) == "2001:db8::1" # IPv6 get_settings.cache_clear() def test_app_refuses_to_start_with_insecure_config(monkeypatch): """Приложение не стартует с небезопасной конфигурацией (например, API_TOKEN=change-me).""" monkeypatch.setenv("API_TOKEN", "change-me") get_settings.cache_clear() with pytest.raises(RuntimeError, match="Небезопасная конфигурация"): with TestClient(create_app()): pass get_settings.cache_clear() def test_login_lockout_by_ip(): """5 неверных попыток входа с одного IP блокируют его на 6-ю (пароль уже не проверяется, код 429); повторные попытки во время блокировки не засоряют журнал новыми auth.locked; блокировка не распространяется на другой IP; успешный вход сбрасывает счётчик неудач.""" with TestClient(create_app()) as c: for _ in range(5): r = c.post("/login", data={"username": "admin", "password": "wrong"}) assert r.status_code == 401 r = c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}) # верный пароль уже не спасает assert r.status_code == 429 and "Слишком много попыток" in r.text assert len(events.list_events(type_="auth.locked")) == 1 for _ in range(3): # попытки во время блокировки — 429, но новых auth.locked не пишут (не засоряют журнал) assert c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}).status_code == 429 assert len(events.list_events(type_="auth.locked")) == 1 with TestClient(create_app(), client=("10.0.0.2", 1)) as c2: r = c2.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}, follow_redirects=False) assert r.status_code == 303 # другой IP свободен with TestClient(create_app(), client=("10.0.0.3", 1)) as c3: for _ in range(3): assert c3.post("/login", data={"username": "admin", "password": "wrong"}).status_code == 401 r = c3.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}, follow_redirects=False) assert r.status_code == 303 for _ in range(4): # если бы счётчик не сбросился успехом, пятая по счёту неудача (2-я в этом цикле) заблокировала бы assert c3.post("/login", data={"username": "admin", "password": "wrong"}).status_code == 401 def test_login_lockout_by_ip_behind_trusted_proxy(monkeypatch): """За доверенным прокси блокировка и событие auth.* считаются по адресу из X-Forwarded-For, а не по адресу прокси: 5 неверных попыток с одним XFF блокируют этот адрес, попытка с другим XFF (тот же прокси-peer) проходит.""" monkeypatch.setenv("TRUSTED_PROXIES", "10.0.0.0/8") get_settings.cache_clear() with TestClient(create_app(), client=("10.0.0.5", 1)) as c: for _ in range(5): r = c.post("/login", data={"username": "admin", "password": "wrong"}, headers={"X-Forwarded-For": "1.1.1.1"}) assert r.status_code == 401 r = c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}, headers={"X-Forwarded-For": "1.1.1.1"}) assert r.status_code == 429 # заблокирован именно реальный адрес клиента assert json.loads(events.list_events(type_="auth.locked")[0].data)["ip"] == "1.1.1.1" r = c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}, headers={"X-Forwarded-For": "2.2.2.2"}, follow_redirects=False) assert r.status_code == 303 # другой адрес за тем же прокси не заблокирован get_settings.cache_clear() def test_move_redirect_rejects_open_redirect_next(): """`next` в /ui/move принимает только локальный путь — иначе редирект на «/» (открытый редирект).""" with TestClient(create_app()) as c: c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD}) for bad in ("/\\evil.com", "//evil.com", "https://evil.com", "/\r\nX"): # последний — декодированный /%0d%0aX r = c.post("/ui/move", data={"next": bad}, follow_redirects=False) assert r.headers["location"] == "/" r = c.post("/ui/move", data={"next": "/?f_group=none"}, follow_redirects=False) assert r.headers["location"] == "/?f_group=none" # обычный путь с фильтром сохраняется def test_process_lock_blocks_second_process(tmp_path): """Второй процесс на той же файловой БД не стартует; после освобождения блокировки — снова можно.""" db_url = f"sqlite:///{tmp_path}/lock.db" process_lock.acquire(db_url) try: with pytest.raises(RuntimeError, match="уже используется"): process_lock.acquire(db_url) finally: process_lock.release() process_lock.acquire(db_url) # после освобождения — успешно process_lock.release()