H3ConuMS-v2/backend/tests/test_security.py

"""Security regression tests for P0 remediations."""
from datetime import datetime, timedelta, timezone

import jwt
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from fastapi import HTTPException
from pydantic import ValidationError
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

from app.core.errors import internal_error
from app.core.credentials import CREDENTIAL_PREFIX, decrypt_credential, encrypt_credential
from app.core.logging_utils import redact_log_message
from app.core.security import verify_casdoor_token
from app.middleware.audit_middleware import sanitize_audit_params


def test_production_rejects_disabled_imc_tls_verification():
    from app.core.config import Settings

    with pytest.raises(ValidationError, match="IMC_API_VERIFY_SSL"):
        Settings(
            SECRET_KEY="test-secret",
            CREDENTIAL_ENCRYPTION_KEY="Q6NgOwoM3sna4ti4UeEo3Lo2cIzxc8wJMvMZ2cmw7lU=",
            DATABASE_URL="sqlite:///./test.db",
            REDIS_URL="redis://localhost:6379/15",
            CASDOOR_ENDPOINT="https://casdoor.example.test",
            CASDOOR_CLIENT_ID="test-client",
            CASDOOR_CLIENT_SECRET="test-client-secret",
            CASDOOR_ORG_NAME="test-org",
            CASDOOR_APP_NAME="test-app",
            IMC_API_URL="https://imc.example.test",
            IMC_API_VERIFY_SSL=False,
            DEBUG=False,
        )


def test_audit_params_redacts_nested_sensitive_values():
    params = {
        "corpsecret": "corp-secret",
        "profile": {"access_token": "access-token", "name": "operator"},
        "items": [{"password": "device-password"}],
        "normal": "kept",
    }

    assert sanitize_audit_params(params) == {
        "corpsecret": "***",
        "profile": {"access_token": "***", "name": "operator"},
        "items": [{"password": "***"}],
        "normal": "kept",
    }


def test_internal_error_does_not_expose_exception_text():
    exception = internal_error("test operation", RuntimeError("database password=should-not-leak"))

    assert isinstance(exception, HTTPException)
    assert exception.status_code == 500
    assert exception.detail["code"] == "INTERNAL_ERROR"
    assert "should-not-leak" not in str(exception.detail)
    assert exception.detail["error_id"]


def test_log_redaction_masks_common_credential_formats():
    message = "Authorization: Bearer abc.def password=hunter2&access_token=token-value corpsecret: corp-secret"
    redacted = redact_log_message(message)

    assert "abc.def" not in redacted
    assert "hunter2" not in redacted
    assert "token-value" not in redacted
    assert "corp-secret" not in redacted


def test_olt_credential_encryption_round_trip(monkeypatch):
    from cryptography.fernet import Fernet
    from app.core.config import settings

    monkeypatch.setattr(settings, "CREDENTIAL_ENCRYPTION_KEY", Fernet.generate_key().decode())
    encrypted = encrypt_credential("olt-device-password")

    assert encrypted.startswith(CREDENTIAL_PREFIX)
    assert "olt-device-password" not in encrypted
    assert decrypt_credential(encrypted) == "olt-device-password"


def test_olt_response_never_contains_device_password():
    from app.schemas.olt import serialize_olt
    from app.models.device import OLTDevice

    device = OLTDevice(
        id=1,
        ip_address="10.0.0.1",
        username="operator",
        password="olt-device-password",
        slot_command="display onu slot",
        region="城区",
    )

    response = serialize_olt(device)
    assert "password" not in response
    assert "olt-device-password" not in str(response)


def test_olt_password_is_encrypted_in_database(monkeypatch):
    from cryptography.fernet import Fernet
    from app.core.config import settings
    from app.core.database import Base
    from app.models.device import OLTDevice

    monkeypatch.setattr(settings, "CREDENTIAL_ENCRYPTION_KEY", Fernet.generate_key().decode())
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    session = sessionmaker(bind=engine)()
    session.add(
        OLTDevice(
            id=1,
            ip_address="10.0.0.1",
            username="operator",
            password="olt-device-password",
            slot_command="display onu slot",
        )
    )
    session.commit()

    stored_password = session.execute(text("SELECT password FROM olt_devices")).scalar_one()
    assert stored_password.startswith(CREDENTIAL_PREFIX)
    assert "olt-device-password" not in stored_password

    session.expire_all()
    assert session.query(OLTDevice).one().password == "olt-device-password"


def test_verify_casdoor_token_requires_expected_signature_and_claims(monkeypatch):
    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    public_pem = private_key.public_key().public_bytes(
        serialization.Encoding.PEM,
        serialization.PublicFormat.SubjectPublicKeyInfo,
    ).decode()
    now = datetime.now(timezone.utc)

    from app.core.config import settings

    monkeypatch.setattr(settings, "CASDOOR_CERTIFICATE", public_pem)
    monkeypatch.setattr(settings, "CASDOOR_ENDPOINT", "https://casdoor.example.test")
    monkeypatch.setattr(settings, "CASDOOR_ISSUER", "")
    monkeypatch.setattr(settings, "CASDOOR_CLIENT_ID", "h3c-client")

    token = jwt.encode(
        {
            "sub": "user-1",
            "iss": "https://casdoor.example.test",
            "aud": "h3c-client",
            "iat": now,
            "exp": now + timedelta(minutes=5),
        },
        private_key,
        algorithm="RS256",
    )

    assert verify_casdoor_token(token)["sub"] == "user-1"

    invalid_token = jwt.encode(
        {
            "sub": "user-1",
            "iss": "https://casdoor.example.test",
            "aud": "other-client",
            "iat": now,
            "exp": now + timedelta(minutes=5),
        },
        private_key,
        algorithm="RS256",
    )
    with pytest.raises(jwt.InvalidAudienceError):
        verify_casdoor_token(invalid_token)