From 99311fdee5c7b76417a7db57de90ed817c61fc3e Mon Sep 17 00:00:00 2001 From: ROHIT-JR Date: Thu, 29 Oct 2026 14:48:28 +0530 Subject: [PATCH] feat(auth): quick-setup helpers and env validation (#70) --- README.md | 24 +++++ nitrostack/__init__.py | 14 +++ nitrostack/auth/setup.py | 221 +++++++++++++++++++++++++++++++++++++++ tests/test_auth_setup.py | 184 ++++++++++++++++++++++++++++++++ 4 files changed, 443 insertions(+) create mode 100644 nitrostack/auth/setup.py create mode 100644 tests/test_auth_setup.py diff --git a/README.md b/README.md index bf299b1..c749128 100644 --- a/README.md +++ b/README.md @@ -294,6 +294,30 @@ ApiKeyModule.for_root(keys=[SecretValue.from_env("SERVICE_API_KEY")]) --- +### Quick auth setup + +One call per auth type returns the keyword arguments for the matching module, and +`validate_auth_env` lists every missing environment variable at once: + +```python +from nitrostack import ( + JWTModule, ApiKeyModule, OAuthModule, + setup_jwt_auth, setup_api_key_auth, setup_oauth_auth, + validate_auth_env, print_auth_setup_instructions, +) + +ok, missing = validate_auth_env("oauth") # (False, ["RESOURCE_URI", "AUTH_SERVER_URL", ...]) +if not ok: + print_auth_setup_instructions("oauth") + +JWTModule.for_root(**setup_jwt_auth()) # reads JWT_SECRET +ApiKeyModule.for_root(**setup_api_key_auth()) # reads API_KEY / API_KEY_ +OAuthModule.for_root(**setup_oauth_auth()) # reads RESOURCE_URI, AUTH_SERVER_URL, JWKS_URI +``` + +`generate_test_credentials()` returns random `SecretValue` credentials for tests and local +demos only. They are never printed; call `.unwrap()` to read one. + ## Transport Options NitroStack apps can run over three transports, selected via `MCP_TRANSPORT_TYPE` (or `ServerConfig(transport_type=...)`): diff --git a/nitrostack/__init__.py b/nitrostack/__init__.py index 4274c09..def2662 100644 --- a/nitrostack/__init__.py +++ b/nitrostack/__init__.py @@ -119,6 +119,14 @@ is_secret_value, unwrap_secret, ) +from nitrostack.auth.setup import ( + generate_test_credentials, + print_auth_setup_instructions, + setup_api_key_auth, + setup_jwt_auth, + setup_oauth_auth, + validate_auth_env, +) from nitrostack.widgets import ( Component, WidgetCsp, @@ -212,6 +220,12 @@ "SecretValue", "is_secret_value", "unwrap_secret", + "setup_jwt_auth", + "setup_api_key_auth", + "setup_oauth_auth", + "generate_test_credentials", + "validate_auth_env", + "print_auth_setup_instructions", "NitroTestingModule", "TaskContext", "LogRecord", diff --git a/nitrostack/auth/setup.py b/nitrostack/auth/setup.py new file mode 100644 index 0000000..d817ada --- /dev/null +++ b/nitrostack/auth/setup.py @@ -0,0 +1,221 @@ +"""One-call setup helpers for JWT, API key and OAuth, plus an environment check. + +Each ``setup_*_auth`` helper returns a plain ``dict`` of keyword arguments. Pass it straight +to the matching module:: + + JWTModule.for_root(**setup_jwt_auth()) + ApiKeyModule.for_root(**setup_api_key_auth()) + OAuthModule.for_root(**setup_oauth_auth()) + +Secrets in the returned dict are ``SecretValue`` objects, so printing or logging the config +never shows them. +""" + +import os +import secrets +from typing import Any, Dict, List, Optional, Sequence, Tuple, Union + +from nitrostack.auth.secret import SecretValue +from nitrostack.core.errors import ConfigurationError + +AUTH_KINDS = ("jwt", "api_key", "oauth") + +# Each entry is a group of alternatives: the group is satisfied if ANY name in it is set. +_REQUIRED_ENV: Dict[str, List[Tuple[str, ...]]] = { + "jwt": [("JWT_SECRET",)], + "api_key": [("API_KEY",)], + "oauth": [ + ("RESOURCE_URI",), + ("AUTH_SERVER_URL",), + ("JWKS_URI", "OAUTH_INTROSPECTION_ENDPOINT", "INTROSPECTION_ENDPOINT"), + ], +} + +_INSTRUCTIONS = { + "jwt": """JWT auth setup + 1. Set a long random secret: JWT_SECRET= + 2. In your module: JWTModule.for_root(**setup_jwt_auth()) + Optional: TOKEN_AUDIENCE, TOKEN_ISSUER""", + "api_key": """API key auth setup + 1. Set one or more keys: API_KEY= (or API_KEY_1, API_KEY_2, ...) + 2. In your module: ApiKeyModule.for_root(**setup_api_key_auth()) + Clients send the key in the x-api-key header.""", + "oauth": """OAuth auth setup + 1. Set the server identity: RESOURCE_URI=https://your-server/mcp + AUTH_SERVER_URL=https://your-authorization-server + 2. Set ONE way to verify tokens: + JWKS_URI=https://your-authorization-server/.well-known/jwks.json + or OAUTH_INTROSPECTION_ENDPOINT=https://your-authorization-server/introspect + 3. In your module: OAuthModule.for_root(**setup_oauth_auth()) + Optional: TOKEN_AUDIENCE, TOKEN_ISSUER, INTROSPECTION_CLIENT_ID, INTROSPECTION_CLIENT_SECRET""", +} + + +def _check_kind(kind: str) -> str: + if kind not in AUTH_KINDS: + raise ValueError(f"Unknown auth kind {kind!r}; expected one of {', '.join(AUTH_KINDS)}") + return kind + + +def _is_set(name: str) -> bool: + return bool((os.environ.get(name) or "").strip()) + + +def _api_key_names() -> List[str]: + return [k for k in os.environ if (k == "API_KEY" or k.startswith("API_KEY_")) and _is_set(k)] + + +def validate_auth_env(kind: str) -> Tuple[bool, List[str]]: + """Check the environment for ``kind`` (``"jwt"``, ``"api_key"`` or ``"oauth"``). + + Returns ``(valid, missing)``. ``missing`` lists every missing variable, not just the + first. When any one of several variables would do, they are shown joined with `` or ``. + """ + _check_kind(kind) + missing: List[str] = [] + for group in _REQUIRED_ENV[kind]: + if kind == "api_key": + satisfied = bool(_api_key_names()) + label = "API_KEY (or API_KEY_)" + else: + satisfied = any(_is_set(name) for name in group) + label = " or ".join(group) + if not satisfied: + missing.append(label) + return (not missing, missing) + + +def print_auth_setup_instructions(kind: str) -> None: + """Print how to configure ``kind``. Never prints any secret value.""" + print(_INSTRUCTIONS[_check_kind(kind)]) + + +def _require_env(kind: str) -> None: + valid, missing = validate_auth_env(kind) + if not valid: + raise ConfigurationError( + f"Missing environment for {kind} auth: {', '.join(missing)}. " + f"Call print_auth_setup_instructions({kind!r}) for help." + ) + + +def setup_jwt_auth( + secret: Union[SecretValue, str, None] = None, + expires_in: str = "24h", + audience: Optional[str] = None, + issuer: Optional[str] = None, + secret_env_var: str = "JWT_SECRET", +) -> Dict[str, Any]: + """Return keyword arguments for ``JWTModule.for_root``. + + Without ``secret``, the secret is read from ``JWT_SECRET`` and a + ``ConfigurationError`` is raised if it is not set. + """ + if secret is None: + if not _is_set(secret_env_var): + raise ConfigurationError( + f"Missing environment for jwt auth: {secret_env_var}. " + "Call print_auth_setup_instructions('jwt') for help." + ) + secret = SecretValue.from_env(secret_env_var) + elif not isinstance(secret, SecretValue): + secret = SecretValue(secret) + return { + "secret_env_var": secret_env_var, + "expires_in": expires_in, + "audience": audience or os.environ.get("TOKEN_AUDIENCE") or None, + "issuer": issuer or os.environ.get("TOKEN_ISSUER") or None, + "secret": secret, + } + + +def setup_api_key_auth( + keys: Optional[Sequence[Union[SecretValue, str]]] = None, + header_name: str = "x-api-key", + hashed: bool = False, + keys_env_prefix: str = "API_KEY", +) -> Dict[str, Any]: + """Return keyword arguments for ``ApiKeyModule.for_root``. + + Without ``keys``, at least one ``API_KEY`` / ``API_KEY_`` variable must be set. + """ + if keys: + wrapped = [k if isinstance(k, SecretValue) else SecretValue(k) for k in keys] + else: + names = [ + k + for k in os.environ + if (k == keys_env_prefix or k.startswith(f"{keys_env_prefix}_")) and _is_set(k) + ] + if not names: + raise ConfigurationError( + f"Missing environment for api_key auth: {keys_env_prefix} " + f"(or {keys_env_prefix}_). " + "Call print_auth_setup_instructions('api_key') for help." + ) + wrapped = [] # the module reads these from the environment itself + return { + "keys_env_prefix": keys_env_prefix, + "header_name": header_name, + "hashed": hashed, + "keys": wrapped, + } + + +def setup_oauth_auth( + resource_uri: Optional[str] = None, + authorization_servers: Optional[Sequence[str]] = None, + scopes_supported: Optional[Sequence[str]] = None, + **overrides: Any, +) -> Dict[str, Any]: + """Return keyword arguments for ``OAuthModule.for_root``. + + Values not passed are read from ``RESOURCE_URI`` and ``AUTH_SERVER_URL``. Token + verification (``JWKS_URI`` or an introspection endpoint) is picked up by the module + from the environment, so one of them must be set unless passed in ``overrides`` + (for example ``jwks_uri=...`` or ``token_introspection_endpoint=...``). + Any other ``OAuthModule.for_root`` argument can be passed as an override. + """ + resource = resource_uri or os.environ.get("RESOURCE_URI") + servers = list(authorization_servers or []) + if not servers and _is_set("AUTH_SERVER_URL"): + servers = [os.environ["AUTH_SERVER_URL"].strip()] + has_verifier = ( + overrides.get("jwks_uri") + or overrides.get("token_introspection_endpoint") + or any(_is_set(n) for n in _REQUIRED_ENV["oauth"][2]) + ) + missing: List[str] = [] + if not resource: + missing.append("RESOURCE_URI") + if not servers: + missing.append("AUTH_SERVER_URL") + if not has_verifier: + missing.append(" or ".join(_REQUIRED_ENV["oauth"][2])) + if missing: + raise ConfigurationError( + f"Missing environment for oauth auth: {', '.join(missing)}. " + "Call print_auth_setup_instructions('oauth') for help." + ) + for name in ("token_introspection_client_secret", "static_client_secret"): + value = overrides.get(name) + if isinstance(value, str): + overrides[name] = SecretValue(value) + return { + "resource_uri": resource, + "authorization_servers": servers, + "scopes_supported": list(scopes_supported or ["read", "write"]), + **overrides, + } + + +def generate_test_credentials() -> Dict[str, SecretValue]: + """Make random credentials for TESTS AND LOCAL DEMOS ONLY. Never use them in production. + + Fresh on every call. The values are ``SecretValue`` objects, so they are not printed + by default; call ``.unwrap()`` to read one. + """ + return { + "jwt_secret": SecretValue(secrets.token_urlsafe(48)), + "api_key": SecretValue(f"sk_test_{secrets.token_hex(24)}"), + } diff --git a/tests/test_auth_setup.py b/tests/test_auth_setup.py new file mode 100644 index 0000000..88bba1a --- /dev/null +++ b/tests/test_auth_setup.py @@ -0,0 +1,184 @@ +"""Issue #70: quick-setup helpers and environment validation for JWT, API key and OAuth.""" + +from __future__ import annotations + +import pytest + +from nitrostack import ( + ApiKeyModule, + DIContainer, + JWTModule, + OAuthModule, + SecretValue, + generate_test_credentials, + print_auth_setup_instructions, + setup_api_key_auth, + setup_jwt_auth, + setup_oauth_auth, + validate_auth_env, +) +from nitrostack.auth.api_key import ApiKeyService +from nitrostack.auth.jwt import JWTService +from nitrostack.auth.oauth import OAuthService +from nitrostack.core.errors import ConfigurationError + +_AUTH_ENV = ( + "JWT_SECRET", "TOKEN_AUDIENCE", "TOKEN_ISSUER", "RESOURCE_URI", "AUTH_SERVER_URL", + "JWKS_URI", "OAUTH_INTROSPECTION_ENDPOINT", "INTROSPECTION_ENDPOINT", +) + + +@pytest.fixture(autouse=True) +def clean(monkeypatch): + import os + + for name in list(os.environ): + if name in _AUTH_ENV or name == "API_KEY" or name.startswith("API_KEY_"): + monkeypatch.delenv(name) + DIContainer.reset() + yield + DIContainer.reset() + + +def test_jwt_config_configures_module(monkeypatch): + monkeypatch.setenv("JWT_SECRET", "x" * 40) + cfg = setup_jwt_auth() + JWTModule.for_root(**cfg) + service = DIContainer.get_instance().resolve(JWTService) + token = service.create_token({"sub": "u1"}) + assert service.verify_token(token)["sub"] == "u1" + assert service.get_secret() == "x" * 40 + + +def test_jwt_explicit_secret_and_plain_str(monkeypatch): + cfg = setup_jwt_auth(secret="y" * 40, audience="aud") + assert isinstance(cfg["secret"], SecretValue) + JWTModule.for_root(**cfg) + service = DIContainer.get_instance().resolve(JWTService) + assert service.audience == "aud" + assert service.get_secret() == "y" * 40 + + +def test_jwt_missing_secret_raises(): + with pytest.raises(ConfigurationError, match="JWT_SECRET"): + setup_jwt_auth() + + +def test_api_key_config_from_explicit_keys(): + cfg = setup_api_key_auth(keys=["k" * 32, SecretValue("m" * 32)]) + ApiKeyModule.for_root(**cfg) + service = DIContainer.get_instance().resolve(ApiKeyService) + assert set(service.get_keys()) == {"k" * 32, "m" * 32} + + +def test_api_key_config_from_env(monkeypatch): + monkeypatch.setenv("API_KEY_1", "a" * 32) + cfg = setup_api_key_auth(header_name="x-token") + assert cfg["keys"] == [] + ApiKeyModule.for_root(**cfg) + service = DIContainer.get_instance().resolve(ApiKeyService) + assert service.get_keys() == ["a" * 32] + assert service.header_name == "x-token" + + +def test_api_key_missing_raises(): + with pytest.raises(ConfigurationError, match="API_KEY"): + setup_api_key_auth() + + +def test_oauth_config_from_env_configures_module(monkeypatch): + monkeypatch.setenv("RESOURCE_URI", "https://srv/mcp") + monkeypatch.setenv("AUTH_SERVER_URL", "https://auth.example") + monkeypatch.setenv("JWKS_URI", "https://auth.example/jwks") + cfg = setup_oauth_auth(scopes_supported=["read"]) + OAuthModule.for_root(**cfg) + service = DIContainer.get_instance().resolve(OAuthService) + assert service.resource_uri == "https://srv/mcp" + assert service.authorization_servers == ["https://auth.example"] + assert service.scopes_supported == ["read"] + assert service.jwks_uri == "https://auth.example/jwks" + + +def test_oauth_overrides_and_secret_wrapping(): + cfg = setup_oauth_auth( + resource_uri="https://srv/mcp", + authorization_servers=["https://auth.example"], + token_introspection_endpoint="https://auth.example/introspect", + token_introspection_client_secret="z" * 32, + ) + assert isinstance(cfg["token_introspection_client_secret"], SecretValue) + OAuthModule.for_root(**cfg) + service = DIContainer.get_instance().resolve(OAuthService) + assert service.token_introspection_endpoint == "https://auth.example/introspect" + + +def test_oauth_reports_every_missing_variable(): + with pytest.raises(ConfigurationError) as err: + setup_oauth_auth() + text = str(err.value) + assert "RESOURCE_URI" in text and "AUTH_SERVER_URL" in text and "JWKS_URI" in text + + +def test_validate_lists_all_missing_not_just_first(): + valid, missing = validate_auth_env("oauth") + assert valid is False + assert len(missing) == 3 + assert missing[0] == "RESOURCE_URI" and missing[1] == "AUTH_SERVER_URL" + assert "JWKS_URI" in missing[2] and "OAUTH_INTROSPECTION_ENDPOINT" in missing[2] + + +def test_validate_partial_and_complete(monkeypatch): + monkeypatch.setenv("RESOURCE_URI", "https://srv/mcp") + assert validate_auth_env("oauth")[1] == [m for m in validate_auth_env("oauth")[1] if "RESOURCE" not in m] + assert len(validate_auth_env("oauth")[1]) == 2 + monkeypatch.setenv("AUTH_SERVER_URL", "https://a") + monkeypatch.setenv("INTROSPECTION_ENDPOINT", "https://a/i") + assert validate_auth_env("oauth") == (True, []) + + +def test_validate_jwt_and_api_key(monkeypatch): + assert validate_auth_env("jwt") == (False, ["JWT_SECRET"]) + monkeypatch.setenv("JWT_SECRET", " ") + assert validate_auth_env("jwt")[0] is False + monkeypatch.setenv("JWT_SECRET", "s" * 32) + assert validate_auth_env("jwt") == (True, []) + assert validate_auth_env("api_key")[0] is False + monkeypatch.setenv("API_KEY_2", "k" * 32) + assert validate_auth_env("api_key") == (True, []) + + +def test_validate_unknown_kind(): + with pytest.raises(ValueError): + validate_auth_env("saml") + + +def test_test_credentials_random_each_call_and_not_printed(capsys, caplog): + first = generate_test_credentials() + second = generate_test_credentials() + assert first["jwt_secret"].unwrap() != second["jwt_secret"].unwrap() + assert first["api_key"].unwrap() != second["api_key"].unwrap() + raw = first["jwt_secret"].unwrap() + assert raw not in repr(first) and raw not in str(first) + assert capsys.readouterr().out == "" + assert raw not in caplog.text + + +def test_test_credentials_configure_modules(): + creds = generate_test_credentials() + JWTModule.for_root(**setup_jwt_auth(secret=creds["jwt_secret"])) + ApiKeyModule.for_root(**setup_api_key_auth(keys=[creds["api_key"]])) + di = DIContainer.get_instance() + assert di.resolve(JWTService).get_secret() == creds["jwt_secret"].unwrap() + assert di.resolve(ApiKeyService).get_keys() == [creds["api_key"].unwrap()] + + +@pytest.mark.parametrize("kind", ["jwt", "api_key", "oauth"]) +def test_print_instructions(kind, capsys): + print_auth_setup_instructions(kind) + out = capsys.readouterr().out + assert "for_root" in out + + +def test_config_repr_hides_secret(monkeypatch): + monkeypatch.setenv("JWT_SECRET", "topsecret-" + "q" * 30) + assert "topsecret" not in repr(setup_jwt_auth())