Nexus/tests/test_security_unit.py

"""P0/P2 security regression unit tests (no live server required)."""

from pathlib import Path
from unittest.mock import AsyncMock, MagicMock

import pytest
from starlette.requests import Request

from server.api.dependencies import check_dangerous_command
from server.application.services.script_jobs import master_callback_url
from server.infrastructure.redis.script_execution_store import _parse_server_ids_json
from server.infrastructure.telegram import sanitize_external_message


def test_dangerous_command_rm_rf_detected():
    warnings = check_dangerous_command("rm -rf /")
    assert warnings


def test_dangerous_command_safe_empty():
    assert check_dangerous_command("echo hello") == []


def test_parse_server_ids_invalid_json():
    assert _parse_server_ids_json("not-json") == []


def test_parse_server_ids_valid():
    assert _parse_server_ids_json("[1, 2]") == [1, 2]


def test_master_callback_url_rejects_http_non_local(monkeypatch):
    monkeypatch.setattr(
        "server.application.services.script_jobs.settings.API_BASE_URL",
        "http://api.example.com",
    )
    with pytest.raises(ValueError, match="HTTPS"):
        master_callback_url()


def test_master_callback_url_allows_localhost_http(monkeypatch):
    monkeypatch.setattr(
        "server.application.services.script_jobs.settings.API_BASE_URL",
        "http://localhost:8600",
    )
    assert master_callback_url().endswith("/api/agent/script-callback")


def test_sanitize_external_message_redacts_mysql_url():
    raw = "connect failed: mysql+pymysql://nexus:SecretPass@db.internal:3306/nexus"
    cleaned = sanitize_external_message(raw)
    assert "SecretPass" not in cleaned
    assert "mysql+pymysql://" not in cleaned
    assert "<connection-url-redacted>" in cleaned


def test_sanitize_external_message_redacts_bearer_token():
    raw = "auth failed Bearer eyJhbGciOiJIUzI1NiJ9.payload.sig"
    cleaned = sanitize_external_message(raw)
    assert "eyJhbGci" not in cleaned
    assert "Bearer <redacted>" in cleaned


def test_sanitize_external_message_truncates_long_text():
    assert len(sanitize_external_message("x" * 500)) <= 240


@pytest.mark.asyncio
async def test_jwt_middleware_blocks_missing_token(monkeypatch):
    from server.api import auth_jwt

    monkeypatch.setattr(auth_jwt, "_INSTALL_MODE", False)

    middleware = auth_jwt.JwtAuthMiddleware(app=MagicMock())
    scope = {
        "type": "http",
        "method": "GET",
        "path": "/api/servers/",
        "headers": [],
        "query_string": b"",
        "client": ("testclient", 50000),
        "server": ("testserver", 80),
        "scheme": "http",
        "root_path": "",
    }
    request = Request(scope)
    call_next = AsyncMock()
    response = await middleware.dispatch(request, call_next)

    assert response.status_code == 401
    call_next.assert_not_called()


@pytest.mark.asyncio
async def test_jwt_middleware_skips_public_login(monkeypatch):
    from server.api import auth_jwt

    monkeypatch.setattr(auth_jwt, "_INSTALL_MODE", False)

    middleware = auth_jwt.JwtAuthMiddleware(app=MagicMock())
    scope = {
        "type": "http",
        "method": "POST",
        "path": "/api/auth/login",
        "headers": [],
        "query_string": b"",
        "client": ("testclient", 50000),
        "server": ("testserver", 80),
        "scheme": "http",
        "root_path": "",
    }
    request = Request(scope)
    call_next = AsyncMock(return_value=MagicMock(status_code=200))
    await middleware.dispatch(request, call_next)
    call_next.assert_called_once()


def test_public_path_install_and_logout():
    from server.api.auth_jwt import _is_public_path

    assert _is_public_path("/api/install/status") is True
    assert _is_public_path("/api/auth/logout") is True


def test_html_pages_do_not_load_external_cdn():
    app_dir = Path(__file__).resolve().parent.parent / "web" / "app"
    offenders = []
    for html in app_dir.glob("*.html"):
        text = html.read_text(encoding="utf-8")
        if "cdn.jsdelivr" in text or "unpkg.com" in text:
            offenders.append(html.name)
    assert offenders == [], f"CDN still referenced in: {offenders}"


def test_vendor_manifest_lists_core_assets():
    manifest_path = Path(__file__).resolve().parent.parent / "web" / "app" / "vendor" / "manifest.json"
    assert manifest_path.is_file()
    import json

    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    for name in ("tailwindcss-browser.js", "alpinejs.min.js"):
        assert name in manifest["files"]
        assert (manifest_path.parent / name).is_file()


def test_install_reject_when_env_exists(monkeypatch, tmp_path):
    from fastapi import HTTPException
    from server.api import install as install_api

    env_file = tmp_path / ".env"
    env_file.write_text("NEXUS_SECRET_KEY=test\n", encoding="utf-8")
    monkeypatch.setattr(install_api, "ENV_FILE", env_file)
    monkeypatch.setattr(install_api, "INSTALL_LOCK", tmp_path / ".install_locked")

    with pytest.raises(HTTPException) as exc:
        install_api._reject_if_installed()
    assert exc.value.status_code == 403