diff --git a/README.md b/README.md index 6ecac48..d6404a1 100644 --- a/README.md +++ b/README.md @@ -126,6 +126,41 @@ land in shell history or process listings. On Linux, binding port 80 needs root or `CAP_NET_BIND_SERVICE`. Without either, use `MFILES_GRPC_TOKEN`. +### Signing in once, not on every run + +By default every run opens the browser, and an MFA vault asks for MFA each time. +`token-cache = "keyring"` remembers the sign-in instead: + +``` +pip install "mfiles-grpc[keyring]" +``` + +```toml +[m-files.tool.grpc] +auth = "sso" +token-cache = "keyring" +``` + +* The first run signs in through the browser as before. When the IdP issues a + refresh token (the vault's scopes include `offline_access`), it is kept in the + operating system's credential store (Windows Credential Manager, macOS + Keychain, or the Secret Service on Linux) under the host and the vault GUID. +* Later runs trade the refresh token for a new token with one call to the IdP. + There is no browser and no MFA, until the IdP expires or revokes the refresh + token. Then the browser opens again. An IdP that rotates refresh tokens is + handled: the new one replaces the old. +* Only the refresh token is stored. It is never written to a file, a log line or + a `repr`. A credential store that cannot be reached is logged and ignored: the + sign-in goes on without remembering. +* **A refresh token is a credential.** Whoever can read it can sign in as you to + that vault without MFA. The credential store protects it with your operating + system login. Leave this off on a shared account or a machine you do not trust. +* `mfiles-grpc forget-token` removes it. A token in `MFILES_GRPC_TOKEN` always + wins, and `auth = "password"` ignores the setting. +* Two runs refreshing at the same moment can race when the IdP rotates refresh + tokens: the loser's token is already used, and its next run falls back to the + browser. The library does not lock across processes. + ## Install ``` @@ -160,6 +195,7 @@ address = "localhost:4443" # connect here instead, e.g. a capturing proxy ca-cert = "proxide_ca.crt" # trust these root certificates (PEM) instead of the system's auth = "sso" # "password" (default) or "sso"; see above sso-token = "access" # optional: "id" or "access" +token-cache = "keyring" # optional: remember the SSO sign-in; see "Signing in once" ``` ### Capturing traffic through a proxy @@ -216,6 +252,7 @@ mfiles-grpc capabilities # anonymous; does the host speak gRPC? mfiles-grpc auth-config # anonymous; the vault's SSO settings mfiles-grpc login # are the credentials good? mfiles-grpc check-session # is the session accepted? +mfiles-grpc forget-token # remove the sign-in token-cache remembers mfiles-grpc structure # object types, classes, custom properties ``` diff --git a/pyproject.toml b/pyproject.toml index 6528ad8..487de67 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,6 +35,10 @@ dependencies = [ ] [project.optional-dependencies] +# token-cache = "keyring" remembers the SSO sign-in in the OS credential store. +keyring = [ + "keyring>=25.0", +] dev = [ "grpcio-tools>=1.84.0", "pytest>=9.0.2", diff --git a/src/mfiles_grpc/__main__.py b/src/mfiles_grpc/__main__.py index c114379..bbb306c 100644 --- a/src/mfiles_grpc/__main__.py +++ b/src/mfiles_grpc/__main__.py @@ -7,6 +7,7 @@ mfiles-grpc auth-config anonymous; shows the vault's SSO (OAuth) settings mfiles-grpc login logs in and out; proves the credentials mfiles-grpc check-session logs in and makes one read that needs the session + mfiles-grpc forget-token removes the sign-in that token-cache remembers mfiles-grpc structure lists object types, classes and property definitions """ @@ -16,7 +17,7 @@ from google.protobuf import json_format -from . import sso, structure +from . import sso, structure, token_cache from .client import Client, SessionNotAccepted from .config import load_settings from .proto import pb @@ -75,6 +76,20 @@ def _check_session(args, settings) -> int: return 0 +def _forget_token(args, settings) -> int: + cache = token_cache.open_cache(settings.token_cache, settings.host, settings.vault) + if cache is None: + print('Nothing is remembered: token-cache is not set in the settings file.') + return 0 + try: + cache.clear() + except token_cache.TokenCacheError as e: + print(f"Not cleared: {e}", file=sys.stderr) + return 1 + print(f"Forgot the remembered sign-in for {settings.host}") + return 0 + + def _structure(args, settings) -> int: with Client.connect(settings) as client: print("Object types:") @@ -102,6 +117,7 @@ def main(argv=None) -> int: commands.add_parser("auth-config").set_defaults(run=_auth_config) commands.add_parser("login").set_defaults(run=_login) commands.add_parser("check-session").set_defaults(run=_check_session) + commands.add_parser("forget-token").set_defaults(run=_forget_token) commands.add_parser("structure").set_defaults(run=_structure) args = parser.parse_args(argv) diff --git a/src/mfiles_grpc/client.py b/src/mfiles_grpc/client.py index 61c3576..b64cc4b 100644 --- a/src/mfiles_grpc/client.py +++ b/src/mfiles_grpc/client.py @@ -20,7 +20,7 @@ import grpc -from . import sso +from . import sso, token_cache from .config import ConnectionSettings from .proto import pb, rpc @@ -129,7 +129,10 @@ def connect(cls, settings: ConnectionSettings) -> "Client": config = sso.discover(client, settings.vault) if settings.sso_token: config = dataclasses.replace(config, use_access_token=settings.sso_token == "access") - token = settings.token or sso.acquire_token(config) + # Without a cache the call is exactly what it was before the cache existed. + cache = token_cache.open_cache(settings.token_cache, settings.host, settings.vault) + cache_args = {} if cache is None else {"cache": cache} + token = settings.token or sso.acquire_token(config, **cache_args) client.log_in_with_token(token, settings.vault, config.plugin_name, config.configuration_scope) else: client.log_in(settings.username, settings.password, settings.vault) diff --git a/src/mfiles_grpc/config.py b/src/mfiles_grpc/config.py index d4f83f8..90fffaf 100644 --- a/src/mfiles_grpc/config.py +++ b/src/mfiles_grpc/config.py @@ -15,6 +15,7 @@ ca-cert = "proxy-ca.pem" # trust these root certificates (PEM) instead of the system's auth = "sso" # "password" (default) or "sso" sso-token = "access" # optional: "id" or "access"; default as the vault's plugin says + token-cache = "keyring" # optional: remember the SSO sign-in in the OS credential store gRPC is served on the REST host. With address set, only the connection goes there: the vault host from rest-api-url is still the name logged in to and the name the server @@ -22,6 +23,10 @@ With auth = "sso", a token in the MFILES_GRPC_TOKEN environment variable is used as it is, instead of signing in through the browser. + +With token-cache = "keyring", the browser is needed only for the first sign-in: the refresh +token is kept in the operating system's credential store and traded for a new token each run. +See token_cache.py for what that means for security. """ import os @@ -30,6 +35,8 @@ from typing import Optional from urllib.parse import urlparse +from .token_cache import CACHE_KINDS + DEFAULT_PORT = 443 AUTH_METHODS = ("password", "sso") SSO_TOKEN_KINDS = ("id", "access") @@ -48,6 +55,7 @@ class ConnectionSettings: auth: str = "password" sso_token: Optional[str] = None token: Optional[str] = None + token_cache: Optional[str] = None def __repr__(self) -> str: # Never let the password or a token reach a log line or a traceback. @@ -57,7 +65,8 @@ def hidden(secret: Optional[str]) -> str: return (f"ConnectionSettings(host={self.host!r}, vault={self.vault!r}, " f"username={self.username!r}, password={hidden(self.password)}, " f"port={self.port}, address={self.address!r}, ca_cert={self.ca_cert!r}, " - f"auth={self.auth!r}, sso_token={self.sso_token!r}, token={hidden(self.token)})") + f"auth={self.auth!r}, sso_token={self.sso_token!r}, token={hidden(self.token)}, " + f"token_cache={self.token_cache!r})") @property def target(self) -> str: @@ -102,6 +111,9 @@ def load_settings(path: str = "client-config.toml") -> ConnectionSettings: sso_token = grpc_section.get("sso-token") if sso_token is not None and sso_token not in SSO_TOKEN_KINDS: raise ValueError(f"sso-token = {sso_token!r}: expected one of {', '.join(SSO_TOKEN_KINDS)}") + token_cache = grpc_section.get("token-cache") + if token_cache is not None and token_cache not in CACHE_KINDS: + raise ValueError(f"token-cache = {token_cache!r}: expected one of {', '.join(CACHE_KINDS)}") if auth == "password" and (not common.get("username") or not common.get("password")): raise ValueError('auth = "password" needs username and password in [m-files.tool.common]') @@ -116,4 +128,5 @@ def load_settings(path: str = "client-config.toml") -> ConnectionSettings: auth=auth, sso_token=sso_token, token=os.environ.get(TOKEN_ENVIRONMENT_VARIABLE) if auth == "sso" else None, + token_cache=token_cache if auth == "sso" else None, ) diff --git a/src/mfiles_grpc/sso.py b/src/mfiles_grpc/sso.py index 7d53ce6..571d4de 100644 --- a/src/mfiles_grpc/sso.py +++ b/src/mfiles_grpc/sso.py @@ -37,6 +37,7 @@ from urllib.parse import parse_qs, urlencode, urlparse, urlunparse from .proto import pb +from .token_cache import TokenCache, TokenCacheError from .values import to_python log = logging.getLogger(__name__) @@ -204,12 +205,19 @@ def _post_form(url: str, form: dict) -> dict: raise SsoError(f"Token endpoint answered {e.code}: {e.read().decode('utf-8', 'replace')}") from e -def exchange_code(config: OAuthConfig, code: str, verifier: str, redirect_uri: str, - post: PostForm = _post_form) -> str: +def _select_token(config: OAuthConfig, tokens: dict) -> str: """ - :param redirect_uri: The one sent in the authorization URL; the IdP compares them + :param tokens: A token endpoint answer :return: The ID token, or the access token when the plugin asks for it """ + wanted = "access_token" if config.use_access_token else "id_token" + if not tokens.get(wanted): + raise SsoError(f"Token endpoint returned no {wanted}; it returned: {', '.join(sorted(tokens))}") + return tokens[wanted] + + +def _exchange_code_for_tokens(config: OAuthConfig, code: str, verifier: str, redirect_uri: str, + post: PostForm) -> dict: form = { "grant_type": "authorization_code", "client_id": config.client_id, @@ -217,13 +225,38 @@ def exchange_code(config: OAuthConfig, code: str, verifier: str, redirect_uri: s "redirect_uri": redirect_uri, "code_verifier": verifier, } + if config.client_secret: + form["client_secret"] = config.client_secret + return post(config.token_endpoint, form) + + +def exchange_code(config: OAuthConfig, code: str, verifier: str, redirect_uri: str, + post: PostForm = _post_form) -> str: + """ + :param redirect_uri: The one sent in the authorization URL; the IdP compares them + :return: The ID token, or the access token when the plugin asks for it + """ + return _select_token(config, _exchange_code_for_tokens(config, code, verifier, redirect_uri, post)) + + +def refresh(config: OAuthConfig, refresh_token: str, post: PostForm = _post_form) -> dict: + """ + Trade a refresh token for new tokens, without a browser. + + :param refresh_token: One the IdP issued for this client + :return: The token endpoint's answer. It carries a new refresh token when the IdP rotates them. + :raises SsoError: The IdP refused the token, or answered without the token the plugin wants + """ + form = { + "grant_type": "refresh_token", + "client_id": config.client_id, + "refresh_token": refresh_token, + } if config.client_secret: form["client_secret"] = config.client_secret tokens = post(config.token_endpoint, form) - wanted = "access_token" if config.use_access_token else "id_token" - if not tokens.get(wanted): - raise SsoError(f"Token endpoint returned no {wanted}; it returned: {', '.join(sorted(tokens))}") - return tokens[wanted] + _select_token(config, tokens) + return tokens def loopback_address(redirect_uri: str) -> tuple[int, str]: @@ -255,14 +288,78 @@ def _listen(port: int, handler) -> list[http.server.HTTPServer]: return servers +def _cached_call(what: str, call, default=None): + """ + :param what: What the call does, for the log + :param call: The credential store call to make + :return: Its result, or default when the store fails. A broken store must not stop a sign-in. + """ + try: + return call() + except TokenCacheError as e: + log.warning("Cannot %s the remembered sign-in: %s", what, e) + return default + + +def _remember(cache: TokenCache, tokens: dict, previous: Optional[str] = None) -> None: + """ + :param tokens: A token endpoint answer + :param previous: The refresh token that was just used, if any + """ + issued = tokens.get("refresh_token") + if issued: + if issued != previous: + if _cached_call("save", lambda: (cache.save(issued), True)[1], default=False): + log.info("Remembered the sign-in%s", " (the IdP rotated the refresh token)" if previous else "") + elif previous is None: + log.info("The IdP issued no refresh token, so this sign-in is not remembered") + + +def _token_from_cache(config: OAuthConfig, cache: TokenCache, post: PostForm) -> Optional[str]: + """ + :return: A token from the remembered refresh token, or None when none is remembered or the IdP + refuses it, in which case it is forgotten + """ + remembered = _cached_call("read", cache.load) + if not remembered: + return None + try: + tokens = refresh(config, remembered, post) + except SsoError as e: + log.info("The remembered sign-in no longer works, signing in again: %s", e) + _cached_call("clear", cache.clear) + return None + log.info("Signed in with the remembered sign-in, without the browser") + _remember(cache, tokens, previous=remembered) + return _select_token(config, tokens) + + def acquire_token(config: OAuthConfig, open_browser: Callable[[str], object] = webbrowser.open, - post: PostForm = _post_form, timeout: float = 300) -> str: + post: PostForm = _post_form, timeout: float = 300, + cache: Optional[TokenCache] = None) -> str: """ - Sign in through the browser. + Get a token for LogIn: from the remembered sign-in when there is one, else through the browser. :param timeout: Seconds to wait for the user to finish signing in + :param cache: Where the refresh token is remembered, or None to sign in through the browser every + time. A sign-in through the browser is remembered when the IdP issues a refresh token. :return: See exchange_code() """ + if cache is not None: + token = _token_from_cache(config, cache, post) + if token: + return token + tokens = _sign_in_with_browser(config, open_browser, post, timeout) + if cache is not None: + _remember(cache, tokens) + return _select_token(config, tokens) + + +def _sign_in_with_browser(config: OAuthConfig, open_browser: Callable[[str], object], + post: PostForm, timeout: float) -> dict: + """ + :return: The token endpoint's answer to the authorization code + """ port, path = loopback_address(config.redirect_uri) redirect_uri = config.redirect_uri verifier, challenge = pkce_pair() @@ -300,4 +397,4 @@ def log_message(self, format, *args): server.shutdown() server.server_close() code = parse_callback(answer["query"], state) - return exchange_code(config, code, verifier, redirect_uri, post=post) + return _exchange_code_for_tokens(config, code, verifier, redirect_uri, post) diff --git a/src/mfiles_grpc/token_cache.py b/src/mfiles_grpc/token_cache.py new file mode 100644 index 0000000..a75eab8 --- /dev/null +++ b/src/mfiles_grpc/token_cache.py @@ -0,0 +1,120 @@ +# vim: autoindent tabstop=4 shiftwidth=4 expandtab softtabstop=4 filetype=python + +""" +Remembering a browser sign-in between runs, so that MFA is asked once and not on every run. + +What is remembered is the OAuth refresh token and nothing else. The access token is short-lived, +and a JWT does not fit where this is kept anyway: Windows Credential Manager takes at most 2560 bytes +per entry, and the keyring package stores text as UTF-16. Each run therefore trades the refresh +token for a fresh access token with one call to the token endpoint, without a browser. + +A refresh token is a credential: whoever holds it can sign in as the user, to this vault, without +MFA, until the IdP expires or revokes it. It is kept in the operating system's credential store +through the keyring package, never in a file, and never in a log line or a repr. Remembering is +opt-in: set token-cache = "keyring" in [m-files.tool.grpc], and install the extra: + + pip install "mfiles-grpc[keyring]" + +`mfiles-grpc forget-token` removes the remembered sign-in. +""" + +import logging +from typing import Optional, Protocol + +log = logging.getLogger(__name__) + +KEYRING_SERVICE = "mfiles-grpc" + +# The values token-cache accepts in the settings file. +CACHE_KINDS = ("keyring",) + + +class TokenCacheError(Exception): + """The credential store could not be read or written.""" + + +class TokenCache(Protocol): + """Where one account's refresh token is kept.""" + + def load(self) -> Optional[str]: + """:return: The remembered refresh token, or None when there is none""" + + def save(self, refresh_token: str) -> None: + """:param refresh_token: The token to remember, replacing any earlier one""" + + def clear(self) -> None: + """Forget the refresh token. Nothing remembered is not an error.""" + + +def entry_name(host: str, vault: str) -> str: + """ + :param host: The vault host + :param vault: The vault GUID, in any case and with or without braces + :return: The name one account's token is kept under. One sign-in serves one vault on one host. + """ + return f"{host.lower()}|{vault.strip().strip('{}').upper()}" + + +class KeyringCache: + """The refresh token of one account, kept in the operating system's credential store.""" + + def __init__(self, entry: str, keyring_module=None): + """ + :param entry: See entry_name() + :param keyring_module: The keyring package, or a stand-in for it in tests + """ + self.entry = entry + self._keyring = keyring_module + + def _backend(self): + if self._keyring is None: + try: + import keyring + except ImportError as e: + raise TokenCacheError( + 'token-cache = "keyring" needs the keyring package: pip install "mfiles-grpc[keyring]"') from e + self._keyring = keyring + return self._keyring + + def load(self) -> Optional[str]: + try: + return self._backend().get_password(KEYRING_SERVICE, self.entry) or None + except TokenCacheError: + raise + except Exception as e: + raise TokenCacheError(f"Cannot read the credential store: {type(e).__name__}") from e + + def save(self, refresh_token: str) -> None: + try: + self._backend().set_password(KEYRING_SERVICE, self.entry, refresh_token) + except TokenCacheError: + raise + except Exception as e: + raise TokenCacheError(f"Cannot write the credential store: {type(e).__name__}") from e + + def clear(self) -> None: + backend = self._backend() + try: + backend.delete_password(KEYRING_SERVICE, self.entry) + except Exception as e: + # Nothing remembered is what was asked for. Telling "not there" from "cannot reach the + # store" is the keyring package's PasswordDeleteError against everything else. + if type(e).__name__ != "PasswordDeleteError": + raise TokenCacheError(f"Cannot clear the credential store: {type(e).__name__}") from e + + def __repr__(self) -> str: + return f"KeyringCache(entry={self.entry!r})" + + +def open_cache(kind: Optional[str], host: str, vault: str) -> Optional[TokenCache]: + """ + :param kind: The token-cache setting: one of CACHE_KINDS, or None for no remembering + :param host: The vault host + :param vault: The vault GUID + :return: The cache to use, or None when sign-in is not remembered + """ + if kind is None: + return None + if kind == "keyring": + return KeyringCache(entry_name(host, vault)) + raise ValueError(f"token-cache = {kind!r}: expected one of {', '.join(CACHE_KINDS)}") diff --git a/tests/test_client.py b/tests/test_client.py index c0bee00..9830d9e 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -202,6 +202,25 @@ def test_connect_with_sso_discovers_signs_in_and_logs_in(): log_in.assert_called_once_with("ID-TOKEN", "{VAULT}", "Entra", "scope-1") +def test_connect_with_a_token_cache_hands_it_to_the_sign_in(): + config = mock.Mock(plugin_name="Entra", configuration_scope="") + with mock.patch.object(sso, "discover", return_value=config), \ + mock.patch.object(sso, "acquire_token", return_value="T") as acquire, \ + mock.patch.object(Client, "log_in_with_token"): + Client.connect(sso_settings(host="Vault.Example", vault="{aaaa}", token_cache="keyring")).close() + cache = acquire.call_args.kwargs["cache"] + assert cache.entry == "vault.example|AAAA" + + +def test_connect_without_a_token_cache_signs_in_exactly_as_before(): + config = mock.Mock(plugin_name="Entra", configuration_scope="") + with mock.patch.object(sso, "discover", return_value=config), \ + mock.patch.object(sso, "acquire_token", return_value="T") as acquire, \ + mock.patch.object(Client, "log_in_with_token"): + Client.connect(sso_settings()).close() + acquire.assert_called_once_with(config) + + def test_connect_with_a_given_token_skips_the_browser(): config = mock.Mock(plugin_name="Entra", configuration_scope="") with mock.patch.object(sso, "discover", return_value=config), \ diff --git a/tests/test_config.py b/tests/test_config.py index 3e35c4e..be9190d 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -120,6 +120,36 @@ def test_unknown_auth_is_an_error(tmp_path): load_settings(str(path)) +def test_the_token_cache_is_off_unless_asked_for(tmp_path, monkeypatch): + monkeypatch.delenv("MFILES_GRPC_TOKEN", raising=False) + path = tmp_path / "client-config.toml" + path.write_text(SSO_CONFIG) + assert load_settings(str(path)).token_cache is None + + +def test_token_cache_is_read_for_sso(tmp_path, monkeypatch): + monkeypatch.delenv("MFILES_GRPC_TOKEN", raising=False) + path = tmp_path / "client-config.toml" + path.write_text(SSO_CONFIG + 'token-cache = "keyring"\n') + s = load_settings(str(path)) + assert s.token_cache == "keyring" + assert "token_cache='keyring'" in repr(s) + + +def test_an_unknown_token_cache_is_an_error(tmp_path): + path = tmp_path / "client-config.toml" + path.write_text(SSO_CONFIG + 'token-cache = "file"\n') + with pytest.raises(ValueError, match="file"): + load_settings(str(path)) + + +def test_password_login_remembers_nothing(tmp_path): + # There is no browser sign-in to remember, so the setting is ignored rather than half-honoured. + path = tmp_path / "client-config.toml" + path.write_text(CONFIG + 'token-cache = "keyring"\n') + assert load_settings(str(path)).token_cache is None + + def test_token_comes_from_the_environment_and_is_hidden(tmp_path, monkeypatch): monkeypatch.setenv("MFILES_GRPC_TOKEN", "eyJsecret-token") path = tmp_path / "client-config.toml" diff --git a/tests/test_sso.py b/tests/test_sso.py index 59ab444..a1caff4 100644 --- a/tests/test_sso.py +++ b/tests/test_sso.py @@ -224,3 +224,151 @@ def browser(url): (_, form), _ = post.call_args assert form["code"] == "the-code" assert form["redirect_uri"] == redirect + + +# --- Remembering the sign-in: the refresh token --------------------------------------------- + +class MemoryCache: + """A TokenCache over one variable; the calls are recorded for the tests to look at.""" + + def __init__(self, stored=None): + self.stored = stored + self.calls = [] + + def load(self): + self.calls.append("load") + return self.stored + + def save(self, refresh_token): + self.calls.append(("save", refresh_token)) + self.stored = refresh_token + + def clear(self): + self.calls.append("clear") + self.stored = None + + +class BrokenCache: + """A credential store that cannot be reached.""" + + def load(self): + raise sso.TokenCacheError("Cannot read the credential store: OSError") + + def save(self, refresh_token): + raise sso.TokenCacheError("Cannot write the credential store: OSError") + + def clear(self): + raise sso.TokenCacheError("Cannot clear the credential store: OSError") + + +def no_browser(url): + raise AssertionError("the browser must not be opened") + + +def browser_that_signs_in(url): + import urllib.request + q = {k: v[0] for k, v in parse_qs(urlparse(url).query).items()} + urllib.request.urlopen(f"{q['redirect_uri']}?code=the-code&state={q['state']}", timeout=5).read() + + +def loopback_config(**overrides): + with socket.socket() as probe: + probe.bind(("127.0.0.1", 0)) + port = probe.getsockname()[1] + return config(RedirectURI=f"http://localhost:{port}/signin-oidc", **overrides) + + +def test_refresh_sends_the_refresh_token_and_returns_the_new_tokens(): + post = mock.Mock(return_value={"id_token": "ID-2", "refresh_token": "R-2"}) + tokens = sso.refresh(config(), "R-1", post=post) + assert tokens == {"id_token": "ID-2", "refresh_token": "R-2"} + (url, form), _ = post.call_args + assert url == TOKEN + assert form == {"grant_type": "refresh_token", "client_id": "client-1", "refresh_token": "R-1"} + + +def test_refresh_sends_the_client_secret_when_the_plugin_has_one(): + post = mock.Mock(return_value={"access_token": "ACCESS"}) + sso.refresh(config(UseAccessTokenInWeb="true", ClientSecret="s3cr3t"), "R-1", post=post) + (_, form), _ = post.call_args + assert form["client_secret"] == "s3cr3t" + + +def test_refresh_without_the_wanted_token_is_an_error(): + post = mock.Mock(return_value={"access_token": "ACCESS"}) + with pytest.raises(sso.SsoError, match="id_token"): + sso.refresh(config(), "R-1", post=post) + + +def test_a_remembered_sign_in_gets_a_token_without_the_browser(): + cache = MemoryCache("R-1") + post = mock.Mock(return_value={"id_token": "ID-2"}) + assert sso.acquire_token(config(), open_browser=no_browser, post=post, cache=cache) == "ID-2" + assert post.call_count == 1 + assert cache.stored == "R-1" + assert ("save", "R-1") not in cache.calls + + +def test_a_rotated_refresh_token_replaces_the_remembered_one(): + # Some IdPs honour a refresh token once and issue the next in its answer. Keeping the old one + # would make the run after this one fail. + cache = MemoryCache("R-1") + post = mock.Mock(return_value={"id_token": "ID-2", "refresh_token": "R-2"}) + sso.acquire_token(config(), open_browser=no_browser, post=post, cache=cache) + assert cache.stored == "R-2" + + +def test_a_refresh_token_the_idp_refuses_is_forgotten_and_the_browser_signs_in(): + cache = MemoryCache("R-old") + + def post(url, form): + if form["grant_type"] == "refresh_token": + raise sso.SsoError("Token endpoint answered 400: invalid_grant") + return {"id_token": "ID-NEW", "refresh_token": "R-new"} + + token = sso.acquire_token(loopback_config(), open_browser=browser_that_signs_in, post=post, + cache=cache, timeout=10) + assert token == "ID-NEW" + assert "clear" in cache.calls + assert cache.stored == "R-new" + + +def test_a_browser_sign_in_is_remembered_when_the_idp_issues_a_refresh_token(): + cache = MemoryCache() + post = mock.Mock(return_value={"id_token": "ID", "refresh_token": "R-1"}) + token = sso.acquire_token(loopback_config(), open_browser=browser_that_signs_in, post=post, + cache=cache, timeout=10) + assert token == "ID" + assert cache.stored == "R-1" + + +def test_a_browser_sign_in_without_a_refresh_token_remembers_nothing(): + cache = MemoryCache() + post = mock.Mock(return_value={"id_token": "ID"}) + sso.acquire_token(loopback_config(), open_browser=browser_that_signs_in, post=post, + cache=cache, timeout=10) + assert cache.stored is None + + +def test_without_a_cache_the_browser_is_used_as_before(): + post = mock.Mock(return_value={"id_token": "ID", "refresh_token": "R-1"}) + token = sso.acquire_token(loopback_config(), open_browser=browser_that_signs_in, post=post, timeout=10) + assert token == "ID" + assert post.call_count == 1 + + +def test_a_broken_credential_store_does_not_stop_the_sign_in(caplog): + post = mock.Mock(return_value={"id_token": "ID", "refresh_token": "R-1"}) + token = sso.acquire_token(loopback_config(), open_browser=browser_that_signs_in, post=post, + cache=BrokenCache(), timeout=10) + assert token == "ID" + assert "Cannot read the remembered sign-in" in caplog.text + assert "Cannot save the remembered sign-in" in caplog.text + + +def test_no_token_reaches_the_log(caplog): + caplog.set_level("DEBUG") + cache = MemoryCache("R-secret-1") + post = mock.Mock(return_value={"id_token": "ID-secret", "refresh_token": "R-secret-2"}) + sso.acquire_token(config(), open_browser=no_browser, post=post, cache=cache) + assert "secret" not in caplog.text diff --git a/tests/test_token_cache.py b/tests/test_token_cache.py new file mode 100644 index 0000000..32fb470 --- /dev/null +++ b/tests/test_token_cache.py @@ -0,0 +1,99 @@ +import sys +from unittest import mock + +import pytest + +from mfiles_grpc import token_cache +from mfiles_grpc.token_cache import KeyringCache, TokenCacheError + + +class PasswordDeleteError(Exception): + """Same name as the keyring package's: that is how "nothing there" is told from a failure.""" + + +class FakeKeyring: + """The three keyring calls the cache makes, over a dict.""" + + def __init__(self): + self.stored = {} + + def get_password(self, service, name): + return self.stored.get((service, name)) + + def set_password(self, service, name, password): + self.stored[(service, name)] = password + + def delete_password(self, service, name): + if (service, name) not in self.stored: + raise PasswordDeleteError("not found") + del self.stored[(service, name)] + + +def cache(fake=None) -> KeyringCache: + return KeyringCache("vault.example|AAAA", fake or FakeKeyring()) + + +def test_entry_name_is_the_same_for_every_spelling_of_the_account(): + assert token_cache.entry_name("Vault.Example", "{aaaa-bbbb}") == "vault.example|AAAA-BBBB" + assert token_cache.entry_name("vault.example", "AAAA-BBBB") == "vault.example|AAAA-BBBB" + + +def test_nothing_remembered_loads_as_none(): + assert cache().load() is None + + +def test_a_saved_refresh_token_is_loaded_and_replaced_by_the_next_one(): + fake = FakeKeyring() + c = cache(fake) + c.save("refresh-1") + assert c.load() == "refresh-1" + c.save("refresh-2") + assert c.load() == "refresh-2" + assert fake.stored == {("mfiles-grpc", "vault.example|AAAA"): "refresh-2"} + + +def test_clear_forgets_the_token_and_forgetting_nothing_is_fine(): + c = cache() + c.save("refresh-1") + c.clear() + assert c.load() is None + c.clear() + + +def test_a_store_failure_is_reported_without_the_stores_own_message(): + # The message could carry the secret; the type name is enough to tell what went wrong. + broken = mock.Mock() + broken.get_password.side_effect = RuntimeError("secret-refresh-token") + broken.set_password.side_effect = RuntimeError("secret-refresh-token") + broken.delete_password.side_effect = RuntimeError("secret-refresh-token") + c = cache(broken) + for call in (c.load, lambda: c.save("x"), c.clear): + with pytest.raises(TokenCacheError) as e: + call() + assert "secret-refresh-token" not in str(e.value) + assert "RuntimeError" in str(e.value) + + +def test_without_the_keyring_package_the_error_says_how_to_get_it(): + with mock.patch.dict(sys.modules, {"keyring": None}): + with pytest.raises(TokenCacheError, match=r"mfiles-grpc\[keyring\]"): + KeyringCache("vault.example|AAAA").load() + + +def test_repr_names_the_entry_and_nothing_else(): + assert repr(cache()) == "KeyringCache(entry='vault.example|AAAA')" + + +def test_open_cache_without_a_kind_remembers_nothing(): + assert token_cache.open_cache(None, "vault.example", "{AAAA}") is None + + +def test_open_cache_keyring_names_the_account(): + c = token_cache.open_cache("keyring", "Vault.Example", "{aaaa}") + assert isinstance(c, KeyringCache) + assert c.entry == "vault.example|AAAA" + + +def test_open_cache_with_an_unknown_kind_is_an_error(): + with pytest.raises(ValueError, match="file"): + token_cache.open_cache("file", "vault.example", "{AAAA}")