diff --git a/README.md b/README.md index 01dcd2a..bf299b1 100644 --- a/README.md +++ b/README.md @@ -266,6 +266,32 @@ The SDK reads standard settings from the environment or `.env` files: | `NITROSTACK_LOG_TO_STDOUT` | Set to `true` to allow logging to stdout under stdio transport (Caution: may corrupt protocol stream). | | `NITROSTACK_HTTP_DEBUG` | Set to `1` to log every HTTP request (method, path, headers, body) and response status to stderr. Use when diagnosing a client that fails to connect: uvicorn's access log shows neither headers nor timestamps. | +### Keeping secrets out of logs (`SecretValue`) + +Wrap a secret in `SecretValue` so it cannot end up in logs, `repr`, tracebacks, pickles or +serialized output. Everything prints `[SecretValue: REDACTED]`. The raw value is only available +through `unwrap()`. + +```python +from nitrostack import SecretValue, JWTModule, ApiKeyModule + +jwt_secret = SecretValue.from_env("JWT_SECRET") # raises if the variable is not set +optional = SecretValue.from_env_optional("MAYBE_SECRET") # None if it is not set + +print(f"{jwt_secret}") # [SecretValue: REDACTED] +JWTModule.for_root(secret=jwt_secret) +ApiKeyModule.for_root(keys=[SecretValue.from_env("SERVICE_API_KEY")]) +``` + +- `SecretValue.from_value("...", allow_hardcoded=True, reason="test")` is for values written in code + (tests). It must be allowed explicitly and it warns. +- Pickling and Pydantic/JSON output are redacted too. A pickled secret loads back with no value, and + `unwrap()` raises instead of returning a fake one. +- `==` compares with `hmac.compare_digest`. +- `JWTModule`, `ApiKeyModule`, `OAuthModule` (the two client secrets) and `ConfigService` + (`defaults`, `get_secret()`) accept a `SecretValue` or a plain `str`. +- `is_secret_value(x)` and `unwrap_secret(x)` help code that takes either. + --- ## Transport Options diff --git a/nitrostack/__init__.py b/nitrostack/__init__.py index 9f1c3d5..4e80cf4 100644 --- a/nitrostack/__init__.py +++ b/nitrostack/__init__.py @@ -113,6 +113,11 @@ ConfigModule, ConfigService, ) +from nitrostack.auth.secret import ( + SecretValue, + is_secret_value, + unwrap_secret, +) from nitrostack.widgets import ( Component, WidgetCsp, @@ -204,6 +209,9 @@ "AudienceMismatchError", "ConfigModule", "ConfigService", + "SecretValue", + "is_secret_value", + "unwrap_secret", "NitroTestingModule", "TaskContext", "TaskCancelledError", diff --git a/nitrostack/auth/api_key.py b/nitrostack/auth/api_key.py index fd0eaf2..6cb98b9 100644 --- a/nitrostack/auth/api_key.py +++ b/nitrostack/auth/api_key.py @@ -1,9 +1,10 @@ import secrets import hashlib import os -from typing import List, Optional +from typing import List, Optional, Sequence, Union from nitrostack.core.module import module from nitrostack.core.di import DIContainer +from nitrostack.auth.secret import SecretValue, unwrap_secret class ApiKeyService: def __init__( @@ -11,10 +12,13 @@ def __init__( keys_env_prefix: str = "API_KEY", header_name: str = "x-api-key", hashed: bool = False, + keys: Optional[Sequence[Union[SecretValue, str]]] = None, ): self.keys_env_prefix = keys_env_prefix self.header_name = header_name self.hashed = hashed + # Keys passed in code are used in addition to the ones found in the environment. + self._explicit_keys: List[Union[SecretValue, str]] = list(keys or []) self._keys: List[str] = [] self._load_keys() @@ -29,7 +33,8 @@ def _load_keys(self) -> None: for k, v in os.environ.items(): if k.startswith(f"{self.keys_env_prefix}_"): keys.append(v) - + + keys.extend(unwrap_secret(k) for k in self._explicit_keys) self._keys = keys def get_keys(self) -> List[str]: @@ -65,11 +70,13 @@ def for_root( keys_env_prefix: str = "API_KEY", header_name: str = "x-api-key", hashed: bool = False, + keys: Optional[Sequence[Union[SecretValue, str]]] = None, ): service = ApiKeyService( keys_env_prefix=keys_env_prefix, header_name=header_name, - hashed=hashed + hashed=hashed, + keys=keys, ) DIContainer.get_instance().register_value(ApiKeyService, service) return cls diff --git a/nitrostack/auth/config.py b/nitrostack/auth/config.py index f97c56d..a521972 100644 --- a/nitrostack/auth/config.py +++ b/nitrostack/auth/config.py @@ -2,6 +2,7 @@ from typing import Dict, Any, Callable, Optional from nitrostack.core.module import module from nitrostack.core.di import DIContainer, injectable +from nitrostack.auth.secret import SecretValue @injectable() class ConfigService: @@ -12,6 +13,8 @@ def __init__( defaults: Optional[Dict[str, Any]] = None, validate_fn: Optional[Callable[[Dict[str, Any]], bool]] = None, ): + # ``defaults`` values may be SecretValue instances. They are kept as they are, so + # printing the config or calling ``get()`` never shows the raw secret. self._config: Dict[str, Any] = {} if defaults: self._config.update(defaults) @@ -52,6 +55,25 @@ def get_or_throw(self, key: str) -> Any: raise KeyError(f"Configuration key '{key}' is required but not found.") return self._config[key] + def get_secret(self, key: str, default: Optional[SecretValue] = None) -> Optional[SecretValue]: + """Return a configuration value as a ``SecretValue`` (or ``default`` when missing). + + A value that is already a ``SecretValue`` is returned as it is. A plain string is + wrapped, so callers can pass it on without it being printed by accident. + """ + value = self._config.get(key) + if value is None or value == "": + return default + if isinstance(value, SecretValue): + return value + return SecretValue(str(value), source="config") + + def get_secret_or_throw(self, key: str) -> SecretValue: + secret = self.get_secret(key) + if secret is None: + raise KeyError(f"Configuration key '{key}' is required but not found.") + return secret + @module(name="ConfigModule") class ConfigModule: @classmethod diff --git a/nitrostack/auth/jwt.py b/nitrostack/auth/jwt.py index 19bd0c7..84c9c5b 100644 --- a/nitrostack/auth/jwt.py +++ b/nitrostack/auth/jwt.py @@ -4,9 +4,10 @@ import hashlib import base64 import os -from typing import Any, Dict, Optional +from typing import Any, Dict, Optional, Union from nitrostack.core.module import module from nitrostack.core.di import DIContainer +from nitrostack.auth.secret import SecretValue, unwrap_secret class JWTService: def __init__( @@ -15,13 +16,18 @@ def __init__( expires_in: str = "24h", audience: Optional[str] = None, issuer: Optional[str] = None, + secret: Union[SecretValue, str, None] = None, ): self.secret_env_var = secret_env_var + # An explicit secret wins over the environment variable. Prefer a SecretValue. + self._secret = secret self.expires_in = expires_in self.audience = audience self.issuer = issuer def get_secret(self) -> str: + if self._secret is not None: + return unwrap_secret(self._secret) secret = os.environ.get(self.secret_env_var) if not secret: # Fallback for dev mode/testing so it doesn't fail hard if not set @@ -116,12 +122,14 @@ def for_root( expires_in: str = "24h", audience: Optional[str] = None, issuer: Optional[str] = None, + secret: Union[SecretValue, str, None] = None, ): service = JWTService( secret_env_var=secret_env_var, expires_in=expires_in, audience=audience, - issuer=issuer + issuer=issuer, + secret=secret, ) DIContainer.get_instance().register_value(JWTService, service) return cls diff --git a/nitrostack/auth/oauth.py b/nitrostack/auth/oauth.py index b6b3358..c5b3095 100644 --- a/nitrostack/auth/oauth.py +++ b/nitrostack/auth/oauth.py @@ -4,11 +4,12 @@ import json import urllib.request import urllib.parse -from typing import List, Optional, Dict, Any, Tuple +from typing import List, Optional, Dict, Any, Tuple, Union from http.server import HTTPServer, BaseHTTPRequestHandler import threading from nitrostack.core.module import module from nitrostack.core.di import DIContainer +from nitrostack.auth.secret import SecretValue, unwrap_secret from nitrostack.auth.oauth_module import ( apply_cimd_to_registration_body, build_authorization_server_metadata, @@ -72,7 +73,7 @@ def __init__( scopes_supported: List[str], token_introspection_endpoint: Optional[str] = None, token_introspection_client_id: Optional[str] = None, - token_introspection_client_secret: Optional[str] = None, + token_introspection_client_secret: Union[SecretValue, str, None] = None, discovery_port: int = 3005, jwks_uri: Optional[str] = None, audience: Optional[str] = None, @@ -80,7 +81,7 @@ def __init__( token_cache_seconds: Optional[int] = None, enable_client_registration: Optional[bool] = None, static_client_id: Optional[str] = None, - static_client_secret: Optional[str] = None, + static_client_secret: Union[SecretValue, str, None] = None, ): self.resource_uri = resource_uri self.authorization_servers = authorization_servers @@ -375,7 +376,8 @@ async def _introspect_via_endpoint(self, token: str) -> Dict[str, Any]: # Add basic auth if client credentials provided if self.token_introspection_client_id and self.token_introspection_client_secret: import base64 - auth_str = f"{self.token_introspection_client_id}:{self.token_introspection_client_secret}" + client_secret = unwrap_secret(self.token_introspection_client_secret) + auth_str = f"{self.token_introspection_client_id}:{client_secret}" encoded_auth = base64.b64encode(auth_str.encode("utf-8")).decode("utf-8") req.add_header("Authorization", f"Basic {encoded_auth}") @@ -460,7 +462,7 @@ def for_root( scopes_supported: List[str], token_introspection_endpoint: Optional[str] = None, token_introspection_client_id: Optional[str] = None, - token_introspection_client_secret: Optional[str] = None, + token_introspection_client_secret: Union[SecretValue, str, None] = None, discovery_port: int = 3005, jwks_uri: Optional[str] = None, audience: Optional[str] = None, @@ -468,7 +470,7 @@ def for_root( token_cache_seconds: Optional[int] = None, enable_client_registration: Optional[bool] = None, static_client_id: Optional[str] = None, - static_client_secret: Optional[str] = None, + static_client_secret: Union[SecretValue, str, None] = None, ): if not resource_uri or not str(resource_uri).strip(): raise ConfigurationError("OAuthModule.for_root requires resource_uri") diff --git a/nitrostack/auth/oauth_module.py b/nitrostack/auth/oauth_module.py index bfc59d1..d0d6176 100644 --- a/nitrostack/auth/oauth_module.py +++ b/nitrostack/auth/oauth_module.py @@ -20,6 +20,8 @@ import time from typing import Any, Dict, Optional, TYPE_CHECKING +from nitrostack.auth.secret import unwrap_secret + from nitrostack.auth.cimd import ( looks_like_cimd_url, resolve_cimd_sync, @@ -153,7 +155,8 @@ def build_registration_response(service: "OAuthService", body: Optional[Dict[str """ body = body or {} client_id = service.static_client_id - client_secret = service.static_client_secret or "" + # The registration response must hand the real secret to the client, so unwrap it here. + client_secret = unwrap_secret(service.static_client_secret) if service.static_client_secret else "" return { "client_id": client_id, "client_secret": client_secret, diff --git a/nitrostack/auth/secret.py b/nitrostack/auth/secret.py new file mode 100644 index 0000000..452afac --- /dev/null +++ b/nitrostack/auth/secret.py @@ -0,0 +1,221 @@ +"""``SecretValue``: a wrapper that keeps a secret out of logs, ``repr``, tracebacks, +pickles and serialized output. + +The raw value is only reachable through :meth:`SecretValue.unwrap`. Everything else +that could print or serialize the object shows ``[SecretValue: REDACTED]`` instead. + +>>> s = SecretValue.from_value("a-long-test-secret-value", allow_hardcoded=True) +>>> f"{s}", repr(s) +('[SecretValue: REDACTED]', '[SecretValue: REDACTED]') +""" + +from __future__ import annotations + +import hmac +import os +import secrets as _secrets +import warnings +from typing import Any, Optional, Union + +REDACTED = "[SecretValue: REDACTED]" + +# Secrets shorter than this still work, but trigger a warning. +MIN_RECOMMENDED_LENGTH = 16 + +# Per-process key so hash() never exposes anything derived from the value alone. +_HASH_KEY = _secrets.token_bytes(32) + + +def _redacted_after_pickle(source: str) -> "SecretValue": + """Rebuild a ``SecretValue`` that carries no value (used when unpickling).""" + return SecretValue._empty(source) + + +class SecretValue: + """Holds a secret string and refuses to reveal it by accident.""" + + __slots__ = ("_value", "source") + + def __init__(self, value: str, source: str = "env") -> None: + if not isinstance(value, str): + raise TypeError("SecretValue needs a str value") + if not value: + raise ValueError("Secret value cannot be empty") + if len(value) < MIN_RECOMMENDED_LENGTH: + warnings.warn( + f"[SecretValue] Secret is shorter than {MIN_RECOMMENDED_LENGTH} characters. " + "Consider using a longer secret.", + UserWarning, + stacklevel=2, + ) + self._value: Optional[str] = value + self.source = source + + # ------------------------------------------------------------------ + # Construction + # ------------------------------------------------------------------ + + @classmethod + def _empty(cls, source: str) -> "SecretValue": + """A secret that was redacted on its way through pickle. It has no value.""" + obj = cls.__new__(cls) + obj._value = None + obj.source = source + return obj + + @classmethod + def from_env(cls, env_var_name: str) -> "SecretValue": + """Read a required secret from an environment variable.""" + value = os.environ.get(env_var_name) + if not value: + raise ValueError( + f"Environment variable {env_var_name} is not set. " + "Set it in your environment or .env file." + ) + return cls(value, "env") + + @classmethod + def from_env_optional(cls, env_var_name: str) -> Optional["SecretValue"]: + """Read a secret from an environment variable, or ``None`` when it is not set.""" + value = os.environ.get(env_var_name) + if not value: + return None + return cls(value, "env") + + @classmethod + def from_value( + cls, + value: str, + allow_hardcoded: bool = False, + reason: Optional[str] = None, + ) -> "SecretValue": + """Wrap a value that is written in code. This must be allowed explicitly.""" + if not allow_hardcoded: + raise ValueError( + "Hardcoded secrets are not allowed. Use SecretValue.from_env() to load " + "secrets from environment variables. If you must use a hardcoded value " + "(for example in a test), pass allow_hardcoded=True." + ) + warnings.warn( + "[SecretValue] Using a hardcoded secret value. " + f"Reason: {reason or 'not specified'}. Do not do this in production.", + UserWarning, + stacklevel=2, + ) + return cls(value, "explicit") + + # ------------------------------------------------------------------ + # Access + # ------------------------------------------------------------------ + + def unwrap(self) -> str: + """Return the raw secret. Call this only where the value is really needed.""" + if self._value is None: + raise ValueError( + "This SecretValue was redacted during serialization and has no value." + ) + return self._value + + def is_from_environment(self) -> bool: + return self.source == "env" + + # ------------------------------------------------------------------ + # Redaction: str, repr, format, f-strings, logging, tracebacks + # ------------------------------------------------------------------ + + def __str__(self) -> str: + return REDACTED + + def __repr__(self) -> str: + return REDACTED + + def __format__(self, format_spec: str) -> str: + return format(REDACTED, format_spec) + + # ------------------------------------------------------------------ + # Comparison + # ------------------------------------------------------------------ + + def __eq__(self, other: object) -> bool: + if isinstance(other, SecretValue): + other_value = other._value + elif isinstance(other, str): + other_value = other + else: + return NotImplemented + if self._value is None or other_value is None: + return False + return hmac.compare_digest(self._value.encode("utf-8"), other_value.encode("utf-8")) + + def __ne__(self, other: object) -> bool: + result = self.__eq__(other) + return result if result is NotImplemented else not result + + def __hash__(self) -> int: + if self._value is None: + return hash(("SecretValue", "redacted")) + return hash(hmac.digest(_HASH_KEY, self._value.encode("utf-8"), "sha256")) + + def __bool__(self) -> bool: + return self._value is not None + + # ------------------------------------------------------------------ + # Pickle and copy + # ------------------------------------------------------------------ + + def __reduce__(self): + # The pickle holds only the source. Loading it gives a SecretValue with no + # value, so a secret can neither leak through a pickle nor be silently + # replaced by a placeholder string. + return (_redacted_after_pickle, (self.source,)) + + def __copy__(self) -> "SecretValue": + return self # immutable, so sharing is safe + + def __deepcopy__(self, memo: Any) -> "SecretValue": + return self + + # ------------------------------------------------------------------ + # Pydantic v2: validation and JSON/dict serialization + # ------------------------------------------------------------------ + + @classmethod + def __get_pydantic_core_schema__(cls, source_type: Any, handler: Any): + from pydantic_core import core_schema + + def validate(value: Any) -> "SecretValue": + if isinstance(value, SecretValue): + return value + if isinstance(value, str): + return cls(value, "explicit") + raise ValueError("expected a SecretValue or a string") + + return core_schema.no_info_plain_validator_function( + validate, + serialization=core_schema.plain_serializer_function_ser_schema( + lambda _value: REDACTED, + return_schema=core_schema.str_schema(), + when_used="always", + ), + ) + + @classmethod + def __get_pydantic_json_schema__(cls, core_schema_: Any, handler: Any): + return {"type": "string", "format": "password", "writeOnly": True} + + +def is_secret_value(value: Any) -> bool: + """True when ``value`` is a ``SecretValue``.""" + return isinstance(value, SecretValue) + + +def unwrap_secret(secret: Union[SecretValue, str]) -> str: + """Return the raw string for a ``SecretValue`` or a plain ``str``. + + Plain strings are accepted so existing code keeps working. + """ + if isinstance(secret, SecretValue): + return secret.unwrap() + if isinstance(secret, str): + return secret + raise TypeError(f"Expected a SecretValue or str, got {type(secret).__name__}") diff --git a/tests/test_secret_value.py b/tests/test_secret_value.py new file mode 100644 index 0000000..5843a95 --- /dev/null +++ b/tests/test_secret_value.py @@ -0,0 +1,434 @@ +"""Issue #69: ``SecretValue`` keeps secrets out of logs, repr, pickles and serialized output.""" + +from __future__ import annotations + +import base64 +import copy +import io +import json +import logging +import pickle +import traceback +import warnings + +import pytest +from pydantic import BaseModel + +from nitrostack import ConfigService, DIContainer, SecretValue, is_secret_value, unwrap_secret +from nitrostack.auth.api_key import ApiKeyModule, ApiKeyService +from nitrostack.auth.jwt import JWTModule, JWTService +from nitrostack.auth.oauth import OAuthModule, OAuthService +from nitrostack.auth.oauth_module import build_registration_response +from nitrostack.auth.secret import REDACTED + +RAW = "s3cr3t-value-that-must-not-leak-0123456789" + + +def _secret(value: str = RAW) -> SecretValue: + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + return SecretValue.from_value(value, allow_hardcoded=True, reason="test") + + +def setup_function() -> None: + DIContainer.reset() + + +def teardown_function() -> None: + DIContainer.reset() + + +# --------------------------------------------------------------------------- +# Construction +# --------------------------------------------------------------------------- + + +def test_from_env_reads_the_variable_and_marks_the_source(monkeypatch): + monkeypatch.setenv("MY_SECRET", RAW) + secret = SecretValue.from_env("MY_SECRET") + assert secret.unwrap() == RAW + assert secret.is_from_environment() is True + + +def test_from_env_raises_when_missing_and_optional_returns_none(monkeypatch): + monkeypatch.delenv("MISSING_SECRET", raising=False) + with pytest.raises(ValueError, match="MISSING_SECRET"): + SecretValue.from_env("MISSING_SECRET") + assert SecretValue.from_env_optional("MISSING_SECRET") is None + monkeypatch.setenv("MISSING_SECRET", "") + assert SecretValue.from_env_optional("MISSING_SECRET") is None + + +def test_the_missing_variable_error_never_contains_a_value(monkeypatch): + monkeypatch.delenv("MISSING_SECRET", raising=False) + with pytest.raises(ValueError) as exc: + SecretValue.from_env("MISSING_SECRET") + assert RAW not in str(exc.value) + + +def test_hardcoded_values_need_explicit_permission(): + with pytest.raises(ValueError, match="allow_hardcoded"): + SecretValue.from_value(RAW) + with pytest.warns(UserWarning, match="hardcoded"): + secret = SecretValue.from_value(RAW, allow_hardcoded=True, reason="unit test") + assert secret.unwrap() == RAW + assert secret.is_from_environment() is False + + +def test_empty_and_non_string_values_are_rejected(): + with pytest.raises(ValueError): + SecretValue("") + with pytest.raises(TypeError): + SecretValue(12345) # type: ignore[arg-type] + + +def test_short_secrets_warn_but_still_work(): + with pytest.warns(UserWarning, match="shorter than 16"): + secret = SecretValue("short") + assert secret.unwrap() == "short" + + +# --------------------------------------------------------------------------- +# Acceptance criteria: the raw value never appears +# --------------------------------------------------------------------------- + + +def test_str_repr_and_fstring_are_redacted(): + secret = _secret() + assert str(secret) == REDACTED + assert repr(secret) == REDACTED + assert f"{secret}" == REDACTED + assert f"{secret!r}" == REDACTED + assert "%s %r" % (secret, secret) == f"{REDACTED} {REDACTED}" + assert "{}".format(secret) == REDACTED + assert RAW not in f"{secret:>40}" + + +def test_containers_holding_a_secret_do_not_leak_in_repr(): + secret = _secret() + assert RAW not in repr({"key": secret}) + assert RAW not in repr([secret]) + assert RAW not in repr((secret,)) + + +def test_there_is_no_instance_dict_to_dump(): + with pytest.raises(TypeError): + vars(_secret()) + + +def test_logging_never_contains_the_raw_value(caplog): + secret = _secret() + logger = logging.getLogger("secret-test") + stream = io.StringIO() + handler = logging.StreamHandler(stream) + handler.setFormatter(logging.Formatter("%(levelname)s %(message)s %(args)s")) + logger.addHandler(handler) + logger.setLevel(logging.DEBUG) + try: + with caplog.at_level(logging.DEBUG, logger="secret-test"): + logger.info("token is %s", secret) + logger.info("token is %r", secret) + logger.info(f"token is {secret}") + logger.info("token is %(s)s", {"s": secret}) + try: + raise RuntimeError(f"failed with {secret}") + except RuntimeError: + logger.exception("boom with %s", secret) + finally: + logger.removeHandler(handler) + assert RAW not in stream.getvalue() + assert RAW not in caplog.text + assert all(RAW not in record.getMessage() for record in caplog.records) + assert REDACTED in caplog.text + + +def test_tracebacks_do_not_contain_the_raw_value(): + secret = _secret() + + def fail(): + local_secret = secret # noqa: F841 - shows up in captured locals + raise RuntimeError(f"bad secret {secret!r}") + + try: + fail() + except RuntimeError as exc: + # capture_locals belongs to TracebackException (format_exception() has no such argument). + captured = traceback.TracebackException(type(exc), exc, exc.__traceback__, capture_locals=True) + text = "".join(captured.format()) + assert RAW not in text + assert REDACTED in text + + +def test_pickle_never_contains_the_raw_value(): + secret = _secret() + for protocol in range(0, pickle.HIGHEST_PROTOCOL + 1): + blob = pickle.dumps(secret, protocol=protocol) + assert RAW.encode() not in blob + assert RAW.encode() not in base64.b64encode(blob) + + +def test_an_unpickled_secret_has_no_value_instead_of_a_fake_one(): + restored = pickle.loads(pickle.dumps(_secret())) + assert is_secret_value(restored) + with pytest.raises(ValueError, match="redacted during serialization"): + restored.unwrap() + assert bool(restored) is False + assert restored != _secret() + + +class Settings(BaseModel): + name: str + api_secret: SecretValue + + +def test_pydantic_serialization_is_redacted(): + model = Settings(name="svc", api_secret=_secret()) + assert RAW not in model.model_dump_json() + assert REDACTED in model.model_dump_json() + assert model.model_dump()["api_secret"] == REDACTED + assert RAW not in json.dumps(model.model_dump()) + assert RAW not in repr(model) + assert RAW not in str(model) + assert RAW not in json.dumps(model.model_json_schema()) + + +def test_pydantic_accepts_a_secret_or_a_string_and_keeps_the_value(): + from_secret = Settings(name="a", api_secret=_secret()) + from_text = Settings(name="b", api_secret=RAW) + assert from_secret.api_secret.unwrap() == RAW + assert from_text.api_secret.unwrap() == RAW + assert isinstance(from_text.api_secret, SecretValue) + with pytest.raises(ValueError): + Settings(name="c", api_secret=123) # type: ignore[arg-type] + + +def test_a_bare_secret_is_not_json_serializable_rather_than_leaking(): + with pytest.raises(TypeError): + json.dumps({"secret": _secret()}) + + +def test_json_default_hook_that_uses_str_still_redacts(): + assert RAW not in json.dumps({"secret": _secret()}, default=str) + + +# --------------------------------------------------------------------------- +# Equality, hashing, copying +# --------------------------------------------------------------------------- + + +def test_equality_compares_values_in_constant_time(monkeypatch): + calls: list[tuple[bytes, bytes]] = [] + real = __import__("hmac").compare_digest + + def spy(a, b): + calls.append((a, b)) + return real(a, b) + + monkeypatch.setattr("nitrostack.auth.secret.hmac.compare_digest", spy) + assert _secret() == _secret() + assert _secret() == RAW + assert _secret() != _secret("another-long-secret-value-xyz") + assert _secret() != "different" + assert len(calls) == 4 + + +def test_equality_with_other_types_is_false(): + assert (_secret() == 123) is False + assert (_secret() != 123) is True + + +def test_equal_secrets_hash_equal_and_the_hash_is_not_the_plain_hash(): + assert hash(_secret()) == hash(_secret()) + assert hash(_secret()) != hash(RAW) + assert len({_secret(), _secret()}) == 1 + + +def test_copy_and_deepcopy_keep_the_value(): + secret = _secret() + assert copy.copy(secret).unwrap() == RAW + nested = copy.deepcopy({"k": [secret]}) + assert nested["k"][0].unwrap() == RAW + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def test_is_secret_value_and_unwrap_secret(): + secret = _secret() + assert is_secret_value(secret) is True + assert is_secret_value(RAW) is False + assert unwrap_secret(secret) == RAW + assert unwrap_secret(RAW) == RAW # plain strings are still accepted + with pytest.raises(TypeError): + unwrap_secret(None) # type: ignore[arg-type] + with pytest.raises(TypeError): + unwrap_secret(123) # type: ignore[arg-type] + + +# --------------------------------------------------------------------------- +# Modules accept a SecretValue and still accept a plain str +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("wrap", [True, False], ids=["SecretValue", "str"]) +def test_jwt_module_accepts_secret_and_str(wrap, monkeypatch): + monkeypatch.delenv("JWT_SECRET", raising=False) + JWTModule.for_root(secret=_secret() if wrap else RAW, audience="mcp") + service = DIContainer.get_instance().resolve(JWTService) + assert service.get_secret() == RAW + token = service.create_token({"sub": "u1"}) + assert service.verify_token(token)["sub"] == "u1" + assert RAW not in token # the secret signs the token, it is not inside it + + +def test_jwt_explicit_secret_wins_over_the_environment(monkeypatch): + monkeypatch.setenv("JWT_SECRET", "from-environment-secret-value") + service = JWTService(secret=_secret()) + assert service.get_secret() == RAW + + +def test_jwt_still_uses_the_environment_when_no_secret_is_given(monkeypatch): + monkeypatch.setenv("JWT_SECRET", "from-environment-secret-value") + assert JWTService().get_secret() == "from-environment-secret-value" + + +def test_jwt_service_repr_does_not_contain_the_secret(): + assert RAW not in repr(JWTService(secret=_secret())) + assert RAW not in str(JWTService(secret=_secret())) + + +@pytest.mark.parametrize("wrap", [True, False], ids=["SecretValue", "str"]) +def test_api_key_module_accepts_secret_and_str(wrap, monkeypatch): + for name in ("API_KEY", "API_KEY_1"): + monkeypatch.delenv(name, raising=False) + ApiKeyModule.for_root(keys=[_secret() if wrap else RAW]) + service = DIContainer.get_instance().resolve(ApiKeyService) + assert service.validate(RAW) is True + assert service.validate("wrong") is False + + +def test_api_key_explicit_keys_add_to_environment_keys(monkeypatch): + monkeypatch.setenv("API_KEY", "env-key-value") + service = ApiKeyService(keys=[_secret()]) + assert service.validate("env-key-value") is True + assert service.validate(RAW) is True + + +def test_api_key_hashed_mode_works_with_a_secret_key(): + service = ApiKeyService(hashed=True, keys=[SecretValue("x" * 16)]) + assert service.validate("anything") is False + hashed = service.hash_key("plain") + assert ApiKeyService(hashed=True, keys=[_secret(hashed)]).validate("plain") is True + + +@pytest.mark.parametrize("wrap", [True, False], ids=["SecretValue", "str"]) +def test_oauth_registration_response_returns_the_real_client_secret(wrap): + service = OAuthService( + resource_uri="http://localhost:3000", + authorization_servers=["http://auth.example.test"], + scopes_supported=["read"], + enable_client_registration=True, + static_client_id="client-1", + static_client_secret=_secret() if wrap else RAW, + ) + body = build_registration_response(service) + assert body["client_secret"] == RAW + assert body["token_endpoint_auth_method"] == "client_secret_post" + if wrap: + assert RAW not in repr(service.static_client_secret) + + +def test_oauth_without_a_client_secret_is_a_public_client(): + service = OAuthService( + resource_uri="http://localhost:3000", + authorization_servers=["http://auth.example.test"], + scopes_supported=["read"], + enable_client_registration=True, + static_client_id="client-1", + ) + body = build_registration_response(service) + assert body["client_secret"] == "" + assert body["token_endpoint_auth_method"] == "none" + + +@pytest.mark.parametrize("wrap", [True, False], ids=["SecretValue", "str"]) +def test_oauth_introspection_sends_the_real_secret_in_basic_auth(wrap, monkeypatch): + monkeypatch.delenv("INTROSPECTION_CLIENT_SECRET", raising=False) + service = OAuthService( + resource_uri="http://localhost:3000", + authorization_servers=["http://auth.example.test"], + scopes_supported=["read"], + token_introspection_endpoint="http://auth.example.test/introspect", + token_introspection_client_id="svc", + token_introspection_client_secret=_secret() if wrap else RAW, + ) + seen = {} + + class FakeResponse: + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def read(self): + return b'{"active": true}' + + def fake_urlopen(request, timeout=0): + seen["auth"] = request.get_header("Authorization") + return FakeResponse() + + monkeypatch.setattr("nitrostack.auth.oauth.urllib.request.urlopen", fake_urlopen) + import asyncio + + result = asyncio.run(service._introspect_via_endpoint("tok")) + assert result == {"active": True} + sent = base64.b64decode(seen["auth"].split(" ", 1)[1]).decode() + assert sent == f"svc:{RAW}" + + +def test_oauth_module_for_root_accepts_secret_values(): + OAuthModule.for_root( + resource_uri="http://localhost:3000", + authorization_servers=["http://auth.example.test"], + scopes_supported=["read"], + token_introspection_client_secret=_secret(), + static_client_secret=_secret(), + ) + service = DIContainer.get_instance().resolve(OAuthService) + assert is_secret_value(service.static_client_secret) + assert RAW not in repr(service.static_client_secret) + + +def test_config_service_keeps_secret_defaults_redacted(tmp_path): + config = ConfigService( + env_file_path=str(tmp_path / "missing.env"), + defaults={"DB_PASSWORD": _secret(), "PLAIN": "x"}, + ) + assert is_secret_value(config.get("DB_PASSWORD")) + assert RAW not in str(config.get("DB_PASSWORD")) + assert config.get_secret("DB_PASSWORD").unwrap() == RAW + + +def test_config_service_get_secret_wraps_plain_strings(tmp_path): + config = ConfigService(env_file_path=str(tmp_path / "missing.env"), defaults={"TOKEN": RAW}) + secret = config.get_secret("TOKEN") + assert is_secret_value(secret) + assert RAW not in f"{secret}" + assert secret.unwrap() == RAW + + +def test_config_service_get_secret_missing_empty_and_required(tmp_path): + config = ConfigService( + env_file_path=str(tmp_path / "missing.env"), + defaults={"EMPTY": ""}, + ignore_env_file=True, + ) + assert config.get_secret("NOPE_NOT_SET_ANYWHERE") is None + assert config.get_secret("EMPTY") is None + fallback = _secret() + assert config.get_secret("NOPE_NOT_SET_ANYWHERE", fallback) is fallback + with pytest.raises(KeyError): + config.get_secret_or_throw("NOPE_NOT_SET_ANYWHERE")