[auth] SecretValue wrapper for secrets

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

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",

nitrostack/auth/api_key.py

@@ -1,20 +1,24 @@
 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__(
         self,
         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

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

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

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,15 +73,15 @@ 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,
         issuer: Optional[str] = None,
         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,15 +462,15 @@ 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,
         issuer: Optional[str] = None,
         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")

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,