ros_control/tests/test_security.py

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()