learningcircuit--local-deep-research/tests/web/auth/test_auth_coverage.py

"""Comprehensive tests for auth-related pure logic.

Covers:
- PasswordValidator (strength validation, requirements)
- AccountLockoutManager (lock/unlock, threshold, expiry, eviction)
- URLValidator (safe URLs, redirect validation, scheme checks, DOI, sanitize)
- sanitize_for_log / strip_control_chars
- SessionManager (create, validate, expire, destroy, cleanup, user queries)
"""

import datetime
import time
from datetime import UTC, timedelta, timezone

import pytest

from local_deep_research.security.account_lockout import AccountLockoutManager
from local_deep_research.security.log_sanitizer import (
    sanitize_for_log,
    strip_control_chars,
)
from local_deep_research.security.password_validator import PasswordValidator
from local_deep_research.security.url_validator import (
    URLValidationError,
    URLValidator,
)
from local_deep_research.web.auth.session_manager import SessionManager


# ======================================================================
# PasswordValidator
# ======================================================================


class TestPasswordValidator:
    """Tests for PasswordValidator.validate_strength and get_requirements."""

    def test_valid_password(self):
        assert PasswordValidator.validate_strength("securepass1") == []

    def test_too_short(self):
        errors = PasswordValidator.validate_strength("abc1")
        assert any("at least 8 characters" in e for e in errors)

    def test_no_lowercase(self):
        errors = PasswordValidator.validate_strength("ABCDEFG1")
        assert any("lowercase" in e for e in errors)

    def test_no_digit(self):
        errors = PasswordValidator.validate_strength("abcdefgh")
        assert any("digit" in e for e in errors)

    def test_empty_password(self):
        errors = PasswordValidator.validate_strength("")
        assert (
            len(errors) >= 2
        )  # too short + no lowercase + no digit (at least 2)

    def test_all_failures(self):
        errors = PasswordValidator.validate_strength("AB")
        assert len(errors) == 3

    def test_exactly_8_chars_valid(self):
        assert PasswordValidator.validate_strength("abcdefg1") == []

    def test_requirements_returns_list(self):
        reqs = PasswordValidator.get_requirements()
        assert isinstance(reqs, list)
        assert len(reqs) >= 3

    def test_requirements_mention_length(self):
        reqs = PasswordValidator.get_requirements()
        assert any("8" in r for r in reqs)

    def test_unicode_letters_do_not_count_as_lowercase(self):
        # Only ASCII lowercase a-z counts
        errors = PasswordValidator.validate_strength("ÉÈÜÖ12345678")
        assert any("lowercase" in e for e in errors)

    def test_digit_anywhere(self):
        assert PasswordValidator.validate_strength("aaaaaaaa1") == []
        assert PasswordValidator.validate_strength("1aaaaaaaa") == []

    def test_special_chars_only(self):
        errors = PasswordValidator.validate_strength("!@#$%^&*")
        assert any("lowercase" in e for e in errors)
        assert any("digit" in e for e in errors)


# ======================================================================
# AccountLockoutManager
# ======================================================================


class TestAccountLockoutManager:
    """Tests for AccountLockoutManager lock/unlock/eviction logic."""

    def _make_manager(self, threshold=3, lockout_minutes=15):
        return AccountLockoutManager(
            threshold=threshold, lockout_minutes=lockout_minutes
        )

    def test_not_locked_initially(self):
        mgr = self._make_manager()
        assert mgr.is_locked("alice") is False

    def test_lock_after_threshold(self):
        mgr = self._make_manager(threshold=3)
        for _ in range(3):
            mgr.record_failure("alice")
        assert mgr.is_locked("alice") is True

    def test_not_locked_below_threshold(self):
        mgr = self._make_manager(threshold=3)
        mgr.record_failure("alice")
        mgr.record_failure("alice")
        assert mgr.is_locked("alice") is False

    def test_success_clears_failures(self):
        mgr = self._make_manager(threshold=3)
        mgr.record_failure("alice")
        mgr.record_failure("alice")
        mgr.record_success("alice")
        # After clearing, need 3 more to lock
        mgr.record_failure("alice")
        mgr.record_failure("alice")
        assert mgr.is_locked("alice") is False

    def test_success_unlocks(self):
        mgr = self._make_manager(threshold=2)
        mgr.record_failure("alice")
        mgr.record_failure("alice")
        assert mgr.is_locked("alice") is True
        mgr.record_success("alice")
        assert mgr.is_locked("alice") is False

    def test_lockout_expires(self):
        mgr = self._make_manager(threshold=1, lockout_minutes=0)
        mgr.record_failure("alice")
        # lockout_minutes=0 means the lock is already in the past
        # Force the lock time to past
        with mgr._lock:
            entry = mgr._state.get("alice")
            if entry and entry.get("locked_until"):
                entry["locked_until"] = datetime.datetime.now(
                    timezone.utc
                ) - timedelta(seconds=1)
        assert mgr.is_locked("alice") is False

    def test_separate_users_independent(self):
        mgr = self._make_manager(threshold=2)
        mgr.record_failure("alice")
        mgr.record_failure("alice")
        assert mgr.is_locked("alice") is True
        assert mgr.is_locked("bob") is False

    def test_record_success_on_unknown_user(self):
        mgr = self._make_manager()
        # Should not raise
        mgr.record_success("nobody")
        assert mgr.is_locked("nobody") is False

    def test_eviction_clears_expired(self):
        mgr = self._make_manager(threshold=1, lockout_minutes=0)
        # Add many entries, all expired
        for i in range(100):
            mgr.record_failure(f"user_{i}")
        # Force all locked_until to past
        with mgr._lock:
            for entry in mgr._state.values():
                if entry.get("locked_until"):
                    entry["locked_until"] = datetime.datetime.now(
                        timezone.utc
                    ) - timedelta(seconds=10)
        # Trigger eviction by exceeding _MAX_STATE_ENTRIES
        mgr._MAX_STATE_ENTRIES = 50  # lower threshold for test
        mgr.record_failure("trigger_evict")
        # After eviction, expired entries should be gone
        with mgr._lock:
            # Only the trigger entry (or fewer) should remain
            assert len(mgr._state) <= 2

    def test_threshold_exactly_at_boundary(self):
        mgr = self._make_manager(threshold=5)
        for _ in range(4):
            mgr.record_failure("alice")
        assert mgr.is_locked("alice") is False
        mgr.record_failure("alice")  # 5th attempt
        assert mgr.is_locked("alice") is True


# ======================================================================
# URLValidator
# ======================================================================


class TestURLValidator:
    """Tests for URLValidator static methods."""

    # --- is_unsafe_scheme ---

    def test_unsafe_javascript(self):
        assert URLValidator.is_unsafe_scheme("javascript:alert(1)") is True

    def test_unsafe_data(self):
        assert (
            URLValidator.is_unsafe_scheme("data:text/html,<h1>hi</h1>") is True
        )

    def test_unsafe_vbscript(self):
        assert URLValidator.is_unsafe_scheme("vbscript:MsgBox") is True

    def test_safe_https(self):
        assert URLValidator.is_unsafe_scheme("https://example.com") is False

    def test_empty_not_unsafe(self):
        assert URLValidator.is_unsafe_scheme("") is False

    def test_case_insensitive_scheme(self):
        assert URLValidator.is_unsafe_scheme("JAVASCRIPT:void(0)") is True

    # --- is_safe_url ---

    def test_safe_https_url(self):
        assert URLValidator.is_safe_url("https://example.com") is True

    def test_safe_http_url(self):
        assert URLValidator.is_safe_url("http://example.com") is True

    def test_ftp_is_safe(self):
        assert URLValidator.is_safe_url("ftp://files.example.com") is True

    def test_empty_not_safe(self):
        assert URLValidator.is_safe_url("") is False

    def test_none_not_safe(self):
        assert URLValidator.is_safe_url(None) is False

    def test_fragment_only_allowed(self):
        assert URLValidator.is_safe_url("#section") is True

    def test_fragment_only_disallowed(self):
        assert (
            URLValidator.is_safe_url("#section", allow_fragments=False) is False
        )

    def test_mailto_default_disallowed(self):
        assert URLValidator.is_safe_url("mailto:user@example.com") is False

    def test_mailto_allowed(self):
        assert (
            URLValidator.is_safe_url(
                "mailto:user@example.com", allow_mailto=True
            )
            is True
        )

    def test_no_scheme_requires_scheme(self):
        assert (
            URLValidator.is_safe_url("example.com", require_scheme=True)
            is False
        )

    def test_no_scheme_not_required(self):
        assert (
            URLValidator.is_safe_url("example.com", require_scheme=False)
            is True
        )

    def test_trusted_domains(self):
        assert (
            URLValidator.is_safe_url(
                "https://evil.com", trusted_domains=["example.com"]
            )
            is False
        )
        assert (
            URLValidator.is_safe_url(
                "https://example.com", trusted_domains=["example.com"]
            )
            is True
        )

    def test_subdomain_trusted(self):
        assert (
            URLValidator.is_safe_url(
                "https://sub.example.com", trusted_domains=["example.com"]
            )
            is True
        )

    def test_double_encoding_suspicious(self):
        # %2520 = double-encoded space
        assert (
            URLValidator.is_safe_url("https://example.com/%2520path") is False
        )

    def test_null_byte_suspicious(self):
        assert URLValidator.is_safe_url("https://example.com/%00path") is False

    # --- sanitize_url ---

    def test_sanitize_adds_scheme(self):
        result = URLValidator.sanitize_url("example.com")
        assert result == "https://example.com"

    def test_sanitize_preserves_https(self):
        result = URLValidator.sanitize_url("https://example.com/path")
        assert result == "https://example.com/path"

    def test_sanitize_rejects_javascript(self):
        assert URLValidator.sanitize_url("javascript:alert(1)") is None

    def test_sanitize_empty(self):
        assert URLValidator.sanitize_url("") is None

    def test_sanitize_custom_default_scheme(self):
        result = URLValidator.sanitize_url("example.com", default_scheme="http")
        assert result is not None
        assert result.startswith("http://")

    # --- is_academic_url ---

    def test_academic_arxiv(self):
        assert (
            URLValidator.is_academic_url("https://arxiv.org/abs/1234") is True
        )

    def test_academic_doi(self):
        assert (
            URLValidator.is_academic_url("https://doi.org/10.1234/test") is True
        )

    def test_academic_subdomain(self):
        assert (
            URLValidator.is_academic_url("https://www.nature.com/articles/123")
            is True
        )

    def test_not_academic(self):
        assert URLValidator.is_academic_url("https://google.com") is False

    def test_academic_empty(self):
        assert URLValidator.is_academic_url("") is False

    # --- extract_doi ---

    def test_extract_doi_from_doi_url(self):
        doi = URLValidator.extract_doi("https://doi.org/10.1234/test.5678")
        assert doi is not None
        assert doi.startswith("10.1234")

    def test_extract_doi_direct(self):
        doi = URLValidator.extract_doi("10.1000/xyz123")
        assert doi is not None

    def test_extract_doi_none(self):
        assert URLValidator.extract_doi("https://example.com") is None

    # --- validate_http_url ---

    def test_validate_http_url_valid(self):
        assert (
            URLValidator.validate_http_url("https://example.com/path") is True
        )

    def test_validate_http_url_no_scheme(self):
        with pytest.raises(URLValidationError):
            URLValidator.validate_http_url("example.com")

    def test_validate_http_url_ftp_rejected(self):
        with pytest.raises(URLValidationError):
            URLValidator.validate_http_url("ftp://example.com")

    def test_validate_http_url_empty(self):
        with pytest.raises(URLValidationError):
            URLValidator.validate_http_url("")

    def test_validate_http_url_none(self):
        with pytest.raises(URLValidationError):
            URLValidator.validate_http_url(None)

    # --- is_safe_redirect_url ---

    def test_redirect_relative_safe(self):
        assert (
            URLValidator.is_safe_redirect_url(
                "/dashboard", "http://localhost:5000/"
            )
            is True
        )

    def test_redirect_same_host_safe(self):
        assert (
            URLValidator.is_safe_redirect_url(
                "http://localhost:5000/page", "http://localhost:5000/"
            )
            is True
        )

    def test_redirect_external_unsafe(self):
        assert (
            URLValidator.is_safe_redirect_url(
                "http://evil.com/steal", "http://localhost:5000/"
            )
            is False
        )

    def test_redirect_empty(self):
        assert (
            URLValidator.is_safe_redirect_url("", "http://localhost:5000/")
            is False
        )

    def test_redirect_crlf_injection(self):
        assert (
            URLValidator.is_safe_redirect_url(
                "/page\r\nSet-Cookie: evil=1", "http://localhost:5000/"
            )
            is False
        )

    def test_redirect_protocol_relative(self):
        assert (
            URLValidator.is_safe_redirect_url(
                "//evil.com/steal", "http://localhost:5000/"
            )
            is False
        )

    def test_redirect_backslash_bypass(self):
        assert (
            URLValidator.is_safe_redirect_url(
                "\\evil.com", "http://localhost:5000/"
            )
            is False
        )

    def test_redirect_null_byte(self):
        assert (
            URLValidator.is_safe_redirect_url(
                "/page\x00evil", "http://localhost:5000/"
            )
            is False
        )

    def test_redirect_path_traversal(self):
        assert (
            URLValidator.is_safe_redirect_url(
                "/../etc/passwd", "http://localhost:5000/"
            )
            is False
        )

    def test_redirect_encoded_crlf(self):
        # %0d%0a encoded
        assert (
            URLValidator.is_safe_redirect_url(
                "/page%0d%0aSet-Cookie: x=1", "http://localhost:5000/"
            )
            is False
        )

    # --- get_safe_redirect_path ---

    def test_safe_redirect_path_valid(self):
        path = URLValidator.get_safe_redirect_path(
            "/dashboard", "http://localhost:5000/"
        )
        assert path == "/dashboard"

    def test_safe_redirect_path_with_query(self):
        path = URLValidator.get_safe_redirect_path(
            "/page?tab=settings", "http://localhost:5000/"
        )
        assert path is not None
        assert "tab=settings" in path

    def test_safe_redirect_path_unsafe(self):
        path = URLValidator.get_safe_redirect_path(
            "http://evil.com/steal", "http://localhost:5000/"
        )
        assert path is None

    def test_safe_redirect_path_empty(self):
        assert (
            URLValidator.get_safe_redirect_path("", "http://localhost:5000/")
            is None
        )


# ======================================================================
# Log Sanitizer
# ======================================================================


class TestLogSanitizer:
    """Tests for sanitize_for_log and strip_control_chars."""

    def test_strip_null_byte(self):
        assert strip_control_chars("hello\x00world") == "helloworld"

    def test_strip_newlines(self):
        assert strip_control_chars("line1\nline2\r") == "line1line2"

    def test_strip_tabs(self):
        assert strip_control_chars("col1\tcol2") == "col1col2"

    def test_preserve_normal_ascii(self):
        s = "Hello World 123 !@#"
        assert strip_control_chars(s) == s

    def test_preserve_unicode_letters(self):
        s = "Héllo Wörld 日本語"
        assert strip_control_chars(s) == s

    def test_strip_zero_width_chars(self):
        assert strip_control_chars("a\u200bb\u200fc") == "abc"

    def test_strip_bom(self):
        assert strip_control_chars("\ufeffHello") == "Hello"

    def test_strip_rlo(self):
        # Right-to-left override
        assert strip_control_chars("test\u202eevil") == "testevil"

    def test_sanitize_truncates(self):
        long_input = "a" * 100
        result = sanitize_for_log(long_input, max_length=50)
        assert len(result) == 50
        assert result.endswith("...")

    def test_sanitize_short_input(self):
        result = sanitize_for_log("hello", max_length=50)
        assert result == "hello"

    def test_sanitize_exact_length(self):
        s = "a" * 50
        result = sanitize_for_log(s, max_length=50)
        assert result == s

    def test_sanitize_strips_and_truncates(self):
        # Control chars stripped first, then truncated
        s = "\x00" + "b" * 100
        result = sanitize_for_log(s, max_length=20)
        assert "\x00" not in result
        assert len(result) == 20

    def test_sanitize_very_small_max_length(self):
        result = sanitize_for_log("abcdef", max_length=3)
        assert len(result) == 3

    def test_sanitize_default_max_length(self):
        long_input = "a" * 200
        result = sanitize_for_log(long_input)
        assert len(result) == 50  # default is 50


# ======================================================================
# SessionManager
# ======================================================================


class TestSessionManager:
    """Tests for SessionManager create/validate/destroy/cleanup logic."""

    def _make_manager(self, session_hours=2, remember_days=30):
        """Create a SessionManager with controlled timeouts."""
        mgr = SessionManager.__new__(SessionManager)
        import threading

        mgr.sessions = {}
        mgr._lock = threading.Lock()
        mgr.session_timeout = timedelta(hours=session_hours)
        mgr.remember_me_timeout = timedelta(days=remember_days)
        return mgr

    def test_create_session_returns_string(self):
        mgr = self._make_manager()
        sid = mgr.create_session("alice")
        assert isinstance(sid, str)
        assert len(sid) > 20  # token_urlsafe(32) is ~43 chars

    def test_create_session_unique(self):
        mgr = self._make_manager()
        s1 = mgr.create_session("alice")
        s2 = mgr.create_session("alice")
        assert s1 != s2

    def test_validate_returns_username(self):
        mgr = self._make_manager()
        sid = mgr.create_session("alice")
        assert mgr.validate_session(sid) == "alice"

    def test_validate_unknown_session(self):
        mgr = self._make_manager()
        assert mgr.validate_session("nonexistent") is None

    def test_validate_updates_last_access(self):
        mgr = self._make_manager()
        sid = mgr.create_session("alice")
        first_access = mgr.sessions[sid]["last_access"]
        time.sleep(0.01)
        mgr.validate_session(sid)
        assert mgr.sessions[sid]["last_access"] > first_access

    def test_expired_session_returns_none(self):
        mgr = self._make_manager(session_hours=1)
        sid = mgr.create_session("alice")
        # Force session to be expired
        with mgr._lock:
            mgr.sessions[sid]["last_access"] = datetime.datetime.now(
                UTC
            ) - timedelta(hours=2)
        assert mgr.validate_session(sid) is None
        # Session should be removed
        assert sid not in mgr.sessions

    def test_remember_me_uses_extended_timeout(self):
        mgr = self._make_manager(session_hours=1, remember_days=30)
        sid = mgr.create_session("alice", remember_me=True)
        # Set last_access to 2 hours ago (would expire normal session)
        with mgr._lock:
            mgr.sessions[sid]["last_access"] = datetime.datetime.now(
                UTC
            ) - timedelta(hours=2)
        # Should still be valid with remember_me
        assert mgr.validate_session(sid) == "alice"

    def test_remember_me_expires_eventually(self):
        mgr = self._make_manager(session_hours=1, remember_days=30)
        sid = mgr.create_session("alice", remember_me=True)
        with mgr._lock:
            mgr.sessions[sid]["last_access"] = datetime.datetime.now(
                UTC
            ) - timedelta(days=31)
        assert mgr.validate_session(sid) is None

    def test_destroy_session(self):
        mgr = self._make_manager()
        sid = mgr.create_session("alice")
        mgr.destroy_session(sid)
        assert mgr.validate_session(sid) is None

    def test_destroy_nonexistent(self):
        mgr = self._make_manager()
        # Should not raise
        mgr.destroy_session("nonexistent")

    def test_cleanup_expired_sessions(self):
        mgr = self._make_manager(session_hours=1)
        sid1 = mgr.create_session("alice")
        sid2 = mgr.create_session("bob")
        # Expire sid1
        with mgr._lock:
            mgr.sessions[sid1]["last_access"] = datetime.datetime.now(
                UTC
            ) - timedelta(hours=2)
        mgr.cleanup_expired_sessions()
        assert sid1 not in mgr.sessions
        assert sid2 in mgr.sessions

    def test_get_active_sessions_count(self):
        mgr = self._make_manager()
        mgr.create_session("alice")
        mgr.create_session("bob")
        assert mgr.get_active_sessions_count() == 2

    def test_get_active_sessions_count_excludes_expired(self):
        mgr = self._make_manager(session_hours=1)
        sid1 = mgr.create_session("alice")
        mgr.create_session("bob")
        with mgr._lock:
            mgr.sessions[sid1]["last_access"] = datetime.datetime.now(
                UTC
            ) - timedelta(hours=2)
        assert mgr.get_active_sessions_count() == 1

    def test_get_user_sessions(self):
        mgr = self._make_manager()
        mgr.create_session("alice")
        mgr.create_session("alice")
        mgr.create_session("bob")
        alice_sessions = mgr.get_user_sessions("alice")
        assert len(alice_sessions) == 2
        for s in alice_sessions:
            assert "session_id" in s
            assert s["session_id"].endswith("...")

    def test_get_user_sessions_empty(self):
        mgr = self._make_manager()
        assert mgr.get_user_sessions("nobody") == []

    def test_get_active_usernames(self):
        mgr = self._make_manager()
        mgr.create_session("alice")
        mgr.create_session("bob")
        mgr.create_session("alice")
        usernames = mgr.get_active_usernames()
        assert usernames == {"alice", "bob"}

    def test_get_active_usernames_excludes_expired(self):
        mgr = self._make_manager(session_hours=1)
        sid = mgr.create_session("alice")
        mgr.create_session("bob")
        with mgr._lock:
            mgr.sessions[sid]["last_access"] = datetime.datetime.now(
                UTC
            ) - timedelta(hours=2)
        assert mgr.get_active_usernames() == {"bob"}

    def test_has_active_sessions_for(self):
        mgr = self._make_manager()
        mgr.create_session("alice")
        assert mgr.has_active_sessions_for("alice") is True
        assert mgr.has_active_sessions_for("bob") is False

    def test_has_active_sessions_for_expired(self):
        mgr = self._make_manager(session_hours=1)
        sid = mgr.create_session("alice")
        with mgr._lock:
            mgr.sessions[sid]["last_access"] = datetime.datetime.now(
                UTC
            ) - timedelta(hours=2)
        assert mgr.has_active_sessions_for("alice") is False

    def test_multiple_users_isolation(self):
        mgr = self._make_manager()
        sid_alice = mgr.create_session("alice")
        sid_bob = mgr.create_session("bob")
        mgr.destroy_session(sid_alice)
        assert mgr.validate_session(sid_bob) == "bob"
        assert mgr.has_active_sessions_for("alice") is False
        assert mgr.has_active_sessions_for("bob") is True