341 lines
13 KiB
Python
341 lines
13 KiB
Python
# -*- 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}"
|