account-manager/tests/test_credentials.py

# -*- coding: utf-8 -*-
"""
Tests for credential operations, secret storage, and reveal.

Note: windows_credential integration tests are skipped by default
(the real Windows Credential Manager requires admin and a real user session).
Secret store logic for windows_credential is tested via mocking.
"""
import json
import os
import pytest


class TestCredentialEnvStorage:
    """credential add-secret env / pick --reveal / SQLite no-secret verification."""

    def test_add_secret_env_saves_ref_not_value(self, clean_db):
        """add-secret with storage=env stores only the env var name, not the secret."""
        from service.account_service import cmd_account_add_secret
        from db.connection import get_conn

        # Set the env var so the command can read it if needed
        os.environ["TEST_DEEPSEEK_KEY"] = "sk-test-1234567890abcdef"

        try:
            cmd_account_add_secret(
                platform_input="deepseek",
                credential_type="api_key",
                label="DeepSeek test key",
                secret_storage="env",
                secret_stdin=False,
                secret_ref="TEST_DEEPSEEK_KEY",
                account_id=None,
                environment="production",
                role="api",
                extra_json_str=None,
            )
        finally:
            del os.environ["TEST_DEEPSEEK_KEY"]

        conn = get_conn()
        cur = conn.cursor()
        cur.execute(
            "SELECT secret_ref, secret_mask, secret_storage FROM credentials WHERE platform_key = ?",
            ("deepseek",),
        )
        row = cur.fetchone()
        assert row is not None, "credential should be inserted"
        secret_ref, secret_mask, secret_storage = row
        assert secret_storage == "env"
        assert secret_ref == "TEST_DEEPSEEK_KEY", "secret_ref should be the env var name"
        # secret_mask should be a masked form of "env:TEST_DEEPSEEK_KEY", NOT the actual secret
        assert "sk-test" not in secret_mask, "secret_mask must not contain the actual secret value"
        assert "****" in secret_mask, "mask should contain ****"
        conn.close()

    def test_credential_pick_reveal_reads_environ(self, clean_db):
        """credential pick --reveal retrieves secret value from os.environ."""
        from service.account_service import cmd_account_add_secret, cmd_credential_pick
        import io, sys

        os.environ["MY_SECRET_KEY"] = "sk-actual-secret-value-abcdefgh"
        try:
            cmd_account_add_secret(
                platform_input="deepseek",
                credential_type="api_key",
                label="Reveal test",
                secret_storage="env",
                secret_stdin=False,
                secret_ref="MY_SECRET_KEY",
                account_id=None,
                environment=None,
                role=None,
                extra_json_str=None,
            )

            # credential pick --reveal
            captured = io.StringIO()
            sys.stdout = captured
            try:
                cmd_credential_pick("deepseek", credential_type="api_key", reveal=True)
            finally:
                sys.stdout = sys.__stdout__

            result = json.loads(captured.getvalue())
            assert result["success"] is True
            assert result["secret"] == "sk-actual-secret-value-abcdefgh"
            assert result["secret_mask"] != result["secret"], "mask should differ from secret"
        finally:
            del os.environ["MY_SECRET_KEY"]

    def test_credential_pick_no_reveal_masks(self, clean_db):
        """credential pick (no --reveal) does NOT include the secret field."""
        from service.account_service import cmd_account_add_secret, cmd_credential_pick
        import io, sys

        os.environ["MASK_TEST_KEY"] = "super-secret-value-xyz"
        try:
            cmd_account_add_secret(
                platform_input="deepseek",
                credential_type="api_key",
                label="Mask test",
                secret_storage="env",
                secret_stdin=False,
                secret_ref="MASK_TEST_KEY",
                account_id=None,
                environment=None,
                role=None,
                extra_json_str=None,
            )

            captured = io.StringIO()
            sys.stdout = captured
            try:
                cmd_credential_pick("deepseek", credential_type="api_key", reveal=False)
            finally:
                sys.stdout = sys.__stdout__

            result = json.loads(captured.getvalue())
            assert result["success"] is True
            assert "secret" not in result, "no secret field when reveal=False"
            assert result["secret_mask"] != "super-secret-value-xyz"
        finally:
            del os.environ["MASK_TEST_KEY"]

    def test_env_missing_returns_error(self, clean_db):
        """credential pick --reveal when env var is missing returns SECRET_ENV_MISSING."""
        from service.account_service import cmd_account_add_secret, cmd_credential_pick
        import io, sys

        # Add credential with a ref to a non-existent env var
        cmd_account_add_secret(
            platform_input="deepseek",
            credential_type="api_key",
            label="Missing env test",
            secret_storage="env",
            secret_stdin=False,
            secret_ref="THIS_ENV_VAR_DOES_NOT_EXIST_AT_ALL",
            account_id=None,
            environment=None,
            role=None,
            extra_json_str=None,
        )

        captured = io.StringIO()
        sys.stdout = captured
        try:
            cmd_credential_pick("deepseek", credential_type="api_key", reveal=True)
        finally:
            sys.stdout = sys.__stdout__

        result = json.loads(captured.getvalue())
        assert result["success"] is False
        assert result["error"]["code"] == "ERROR:SECRET_ENV_MISSING"


class TestCredentialNoneStorage:
    """credential add-secret storage=none (browser_profile / note)."""

    def test_add_secret_none(self, clean_db):
        """storage=none works for browser_profile credential type."""
        from service.account_service import cmd_account_add_secret

        cmd_account_add_secret(
            platform_input="maersk",
            credential_type="browser_profile",
            label="Maersk browser session",
            secret_storage="none",
            secret_stdin=False,
            secret_ref=None,
            account_id=None,
            environment="simulator",
            role="booking",
            extra_json_str=None,
        )

        from db.connection import get_conn
        conn = get_conn()
        cur = conn.cursor()
        cur.execute(
            "SELECT secret_storage, secret_ref, secret_mask FROM credentials WHERE platform_key = ?",
            ("maersk",),
        )
        row = cur.fetchone()
        assert row is not None
        assert row[0] == "none"
        assert row[1] is None or row[1] == ""
        conn.close()


class TestCredentialWindowsCredentialMocked:
    """Mocked tests for windows_credential secret store logic."""

    def test_mask_secret_never_returns_full_value(self):
        """mask_secret must never return the actual secret value."""
        from service.secret_store import mask_secret

        secrets = [
            "sk-test-1234567890abcdefghijklmnop",
            "short",
            "abc",
            "a" * 100,
            "",
            None,
        ]
        for s in secrets:
            if s is None or s == "":
                masked = mask_secret(s)
                assert masked == "****"
                continue
            result = mask_secret(s)
            # The mask should never equal the original
            assert result != s, f"mask_secret({s!r}) must not equal original"
            # The mask should always contain "****"
            assert "****" in result, f"mask should contain '****': {result}"
            # For long secrets, original should not appear in mask
            assert s not in result, f"original secret must not appear in mask: {s}"

    def test_read_secret_env_success(self):
        """read_secret with storage=env returns the os.environ value."""
        os.environ["TEST_ENV_FOR_READ"] = "env-secret-value-xyz"
        try:
            from service.secret_store import read_secret
            val = read_secret("env", "TEST_ENV_FOR_READ")
            assert val == "env-secret-value-xyz"
        finally:
            del os.environ["TEST_ENV_FOR_READ"]

    def test_read_secret_env_missing_raises(self):
        """read_secret with storage=env raises ValueError when env var is missing."""
        # Ensure the env var is not set
        os.environ.pop("NONEXISTENT_VAR_FOR_TEST_123", None)
        from service.secret_store import read_secret
        with pytest.raises(ValueError) as exc_info:
            read_secret("env", "NONEXISTENT_VAR_FOR_TEST_123")
        assert "SECRET_ENV_MISSING" in str(exc_info.value)

    def test_read_secret_none(self):
        """read_secret with storage=none returns empty string."""
        from service.secret_store import read_secret
        val = read_secret("none", "ignored-ref")
        assert val == ""

    def test_write_secret_none_is_noop(self):
        """write_secret with storage=none does not raise."""
        from service.secret_store import write_secret
        # Should not raise
        write_secret("none", "ignored-ref", "any-secret-value")

    def test_windows_credential_write_fails_on_non_win32(self, monkeypatch):
        """On non-Windows, windows_credential write raises OSError."""
        import sys
        monkeypatch.setattr(sys, "platform", "linux")
        from service.secret_store import write_secret
        with pytest.raises(OSError) as exc_info:
            write_secret("windows_credential", "some-target", "some-secret")
        assert "WINDOWS_CREDENTIAL_UNAVAILABLE" in str(exc_info.value) or \
               "not available" in str(exc_info.value).lower()


class TestCredentialPickFiltering:
    """credential pick filtering by credential_type / environment."""

    def test_pick_by_credential_type(self, clean_db):
        from service.account_service import cmd_account_add_secret, cmd_credential_pick
        import io, sys

        os.environ["CREDS_TYPE_KEY_A"] = "val-a"
        os.environ["CREDS_TYPE_KEY_B"] = "val-b"
        try:
            cmd_account_add_secret(
                "deepseek", "api_key", "A", "env", False, "CREDS_TYPE_KEY_A",
                None, None, None, None,
            )
            cmd_account_add_secret(
                "deepseek", "model_param", "B", "env", False, "CREDS_TYPE_KEY_B",
                None, None, None, None,
            )

            captured = io.StringIO()
            sys.stdout = captured
            try:
                cmd_credential_pick("deepseek", credential_type="model_param", reveal=False)
            finally:
                sys.stdout = sys.__stdout__

            result = json.loads(captured.getvalue())
            assert result["success"] is True
            assert result["credential_type"] == "model_param"
        finally:
            del os.environ["CREDS_TYPE_KEY_A"]
            del os.environ["CREDS_TYPE_KEY_B"]

    def test_pick_no_credential_returns_error(self, clean_db):
        """credential pick when none matches returns CREDENTIAL_NOT_FOUND."""
        from service.account_service import cmd_credential_pick
        import io, sys

        captured = io.StringIO()
        sys.stdout = captured
        try:
            cmd_credential_pick("deepseek", credential_type="nonexistent_type", reveal=False)
        finally:
            sys.stdout = sys.__stdout__

        result = json.loads(captured.getvalue())
        assert result["success"] is False
        assert result["error"]["code"] == "ERROR:CREDENTIAL_NOT_FOUND"


class TestSQLiteNoPlaintextSecret:
    """Verify that SQLite never stores actual secret values."""

    def test_credentials_table_never_has_plaintext(self, clean_db):
        """All credential rows should have secret_mask != original secret value."""
        from service.account_service import cmd_account_add_secret
        from db.connection import get_conn

        test_secret = "SQLITE_NO_PLAINTEXT_TEST_SECRET_987654321"
        os.environ["SQLITE_NO_PLAINTEXT_TEST"] = test_secret
        try:
            cmd_account_add_secret(
                "deepseek", "api_key", "SQLite test", "env", False,
                "SQLITE_NO_PLAINTEXT_TEST", None, None, None, None,
            )
        finally:
            del os.environ["SQLITE_NO_PLAINTEXT_TEST"]

        conn = get_conn()
        cur = conn.cursor()
        cur.execute("SELECT secret_ref, secret_mask FROM credentials")
        rows = cur.fetchall()
        conn.close()

        for secret_ref, secret_mask in rows:
            # secret_ref for env storage is the env var name, not the secret
            if secret_ref == "SQLITE_NO_PLAINTEXT_TEST":
                # If it matches our test env var name, check the mask
                assert test_secret not in (secret_mask or ""), \
                    f"Plaintext secret must not appear in secret_mask or secret_ref columns. mask={secret_mask}"