From 0cac30ccf896df7b7798191fcba639958e82d73d Mon Sep 17 00:00:00 2001 From: Yoshihiro OKUMURA Date: Fri, 14 Aug 2026 16:36:50 +0900 Subject: [PATCH] fix(auth): serialise the token refresh across processes The login cache is shared by every mdrs process, but the refresh was guarded by a lock that only reaches inside one. Concurrent runs each sent the same refresh token, and a provider that rotates them accepts the first and refuses the rest. - hold a lock that spans processes across the whole read-refresh- write, checking cheaply first so ordinary requests never take it - write the cache through a temporary file: opening it for writing truncates it, and a reader landing in that window found it empty and threw the session away - take the lock for every write, not just the refresh, so a login running beside one cannot be silently reverted - wait for a busy lock on Windows rather than giving up after the ten attempts msvcrt allows, and retry the rename it refuses while a reader still holds the file open - bound the refresh request on its own, so a provider that goes quiet cannot hold the lock indefinitely - accept bare hostnames such as localhost, store URLs without the trailing slash, and join download paths through one helper --- CHANGELOG.md | 12 ++++ mdrsclient/api/users.py | 8 ++- mdrsclient/api/utils.py | 43 +++++++++---- mdrsclient/cache.py | 128 ++++++++++++++++++++++++++++++++------ mdrsclient/commands/ls.py | 3 +- mdrsclient/config.py | 31 +++++++-- mdrsclient/connection.py | 14 ++++- mdrsclient/utils.py | 18 +++++- tests/test_token_cache.py | 116 ++++++++++++++++++++++++++++++++++ 9 files changed, 332 insertions(+), 41 deletions(-) create mode 100644 tests/test_token_cache.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 9d63a4a..a4ef5cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,18 @@ All notable changes to this project will be documented in this file. +## [Unreleased] + +### Fixed +- Serialised the token refresh across processes. Concurrent `mdrs` invocations shared one refresh token and each sent it, so a provider that rotates refresh tokens accepted the first and refused the rest. +- Wrote the login cache through a temporary file so a reader can no longer catch it mid-truncation and discard the session. +- Joined the base URL and the API's relative path correctly in `ls --json` output, which produced a doubled separator when the configured URL ended with one. + +### Changed +- Accepted bare hostnames such as `localhost` in `config create`/`config update`, and stored the URL without its trailing slash, matching the Rust client so both can share `config.ini`. +- Bounded the token refresh request with its own timeout, so a provider that stops answering cannot hold the cross-process lock indefinitely. +- **Breaking for embedders:** `CacheInterface` now requires `lock()` and `reload()`. A cache passed to `MdrsClient.from_remote(..., cache=...)` must provide both; `InMemoryCache` implements them as no-ops. + ## [1.3.18] - 2026-07-02 ### Added diff --git a/mdrsclient/api/users.py b/mdrsclient/api/users.py index 3958e78..9054c2a 100644 --- a/mdrsclient/api/users.py +++ b/mdrsclient/api/users.py @@ -8,6 +8,9 @@ from mdrsclient.api.base import BaseApi from mdrsclient.exceptions import UnauthorizedException from mdrsclient.models import Token, User +# (connect, read) seconds for the token refresh. +TOKEN_REFRESH_TIMEOUT: Final[tuple[float, float]] = (5.0, 30.0) + @dataclass(frozen=True) class UsersCurrentResponseLaboratory: @@ -61,7 +64,10 @@ class UsersApi(BaseApi): # print(self.__class__.__name__ + "::" + sys._getframe().f_code.co_name) url = self.ENTRYPOINT + "token/refresh/" data: dict[str, str | int] = {"refresh": token.refresh} - response = self.connection.post(url, data=data) + # Bounded on its own: the caller holds a lock that spans processes while this runs, + # so a provider that accepts the connection and then goes quiet would otherwise + # stall every other request on this machine rather than just this one. + response = self.connection.post(url, data=data, timeout=TOKEN_REFRESH_TIMEOUT) if response.status_code == requests.codes.unauthorized: raise UnauthorizedException("Token is invalid or expired.") self._raise_response_error(response) diff --git a/mdrsclient/api/utils.py b/mdrsclient/api/utils.py index 6849304..b14e9e2 100644 --- a/mdrsclient/api/utils.py +++ b/mdrsclient/api/utils.py @@ -4,16 +4,35 @@ from mdrsclient.exceptions import UnauthorizedException def token_check(connection: MDRSConnection) -> None: - try: - connection.lock.acquire() - if connection.token is not None: - if connection.token.is_refresh_required: - user_api = UsersApi(connection) - try: - connection.token = user_api.tokenRefresh(connection.token) - except UnauthorizedException: - connection.logout() - elif connection.token.is_expired: + """ + Bring the access token up to date before a request goes out. + + Refreshing is a read-modify-write over a cache shared with every other client + process on this machine, and a rotating provider stops honouring the refresh token + it replaces. Two processes reaching this at once would otherwise both send the same + token, and the loser would be left holding one the server no longer accepts, so the + whole sequence runs under a lock that spans processes and the cache is re-read + inside it. + """ + with connection.lock: + token = connection.token + if token is None or not (token.is_refresh_required or token.is_expired): + # Nothing to do, which is the answer for almost every request. The lock below + # reaches across processes and is held for a round trip, so it is worth + # knowing that before taking it. + return + with connection.cache_lock(): + connection.reload_cache() + token = connection.token + if token is None: + return + if token.is_expired: + connection.logout() + return + if not token.is_refresh_required: + return + user_api = UsersApi(connection) + try: + connection.token = user_api.tokenRefresh(token) + except UnauthorizedException: connection.logout() - finally: - connection.lock.release() diff --git a/mdrsclient/cache.py b/mdrsclient/cache.py index 758ca96..fa6bc4a 100644 --- a/mdrsclient/cache.py +++ b/mdrsclient/cache.py @@ -1,8 +1,12 @@ +import contextlib import dataclasses import hashlib import json import os -from typing import Protocol, runtime_checkable +import tempfile +import time +from contextlib import AbstractContextManager +from typing import Iterator, Protocol, runtime_checkable from pydantic import TypeAdapter, ValidationError from pydantic.dataclasses import dataclass @@ -46,6 +50,19 @@ class CacheData: @runtime_checkable class CacheInterface(Protocol): + def lock(self) -> AbstractContextManager[None]: + """ + Hold exclusive access to the cache for the duration of the block. + + Refreshing a token is a read-modify-write, and the cache is shared with every + other client process using the same remote. + """ + ... + + def reload(self) -> None: + """Re-read the cache, discarding anything held from an earlier read.""" + ... + @property def token(self) -> Token | None: ... @token.setter @@ -70,6 +87,14 @@ class InMemoryCache(CacheInterface): def __init__(self) -> None: self.__data = CacheData() + @contextlib.contextmanager + def lock(self) -> Iterator[None]: + # Nothing else can reach this cache, so there is nothing to exclude. + yield + + def reload(self) -> None: + pass + @property def token(self) -> Token | None: return self.__data.token @@ -109,14 +134,46 @@ class CacheFile(CacheInterface): __serial: int __cache_dir: str __cache_file: str + __lock_file: str + __lock_depth: int __data: CacheData def __init__(self, remote: str) -> None: self.__serial = -1 self.__cache_dir = os.path.join(CONFIG_DIRNAME, "cache") self.__cache_file = os.path.join(self.__cache_dir, remote + ".json") + self.__lock_file = os.path.join(self.__cache_dir, remote + ".lock") + self.__lock_depth = 0 self.__data = CacheData() + @contextlib.contextmanager + def lock(self) -> Iterator[None]: + # Re-entrant, because every write takes it and a refresh is a write made while + # already holding it. A second flock on the same file from the same process would + # wait for a release that cannot come. + if self.__lock_depth > 0: + self.__lock_depth += 1 + try: + yield + finally: + self.__lock_depth -= 1 + return + # A separate file, so that replacing the cache cannot disturb the lock holders. + self.__ensure_cache_dir() + with open(self.__lock_file, "a") as f: + FileLock.lock(f) + self.__lock_depth = 1 + try: + yield + finally: + self.__lock_depth = 0 + FileLock.unlock(f) + os.chmod(self.__lock_file, 0o600) + + def reload(self) -> None: + self.__serial = -1 + self.__load() + @property def token(self) -> Token | None: self.__load() @@ -124,9 +181,10 @@ class CacheFile(CacheInterface): @token.setter def token(self, token: Token) -> None: - self.__load() - self.__data.token = token - self.__save() + with self.lock(): + self.reload() + self.__data.token = token + self.__save() @token.deleter def token(self) -> None: @@ -139,9 +197,10 @@ class CacheFile(CacheInterface): @user.setter def user(self, user: User) -> None: - self.__load() - self.__data.user = user - self.__save() + with self.lock(): + self.reload() + self.__data.user = user + self.__save() @user.deleter def user(self) -> None: @@ -154,9 +213,10 @@ class CacheFile(CacheInterface): @laboratories.setter def laboratories(self, laboratories: Laboratories) -> None: - self.__load() - self.__data.laboratories = laboratories - self.__save() + with self.lock(): + self.reload() + self.__data.laboratories = laboratories + self.__save() def __clear(self) -> None: self.__data.clear() @@ -164,8 +224,7 @@ class CacheFile(CacheInterface): def __load(self) -> None: if os.path.isfile(self.__cache_file): - stat = os.stat(self.__cache_file) - serial = hash((stat.st_uid, stat.st_gid, stat.st_mode, stat.st_size, stat.st_mtime)) + serial = self.__stat_serial() if self.__serial != serial: try: with open(self.__cache_file) as f: @@ -185,15 +244,44 @@ class CacheFile(CacheInterface): def __save(self) -> None: self.__ensure_cache_dir() - with open(self.__cache_file, "w") as f: - FileLock.lock(f) - self.__data.update_digest() - f.write(json.dumps(dataclasses.asdict(self.__data))) - FileLock.unlock(f) + self.__data.update_digest() + payload = json.dumps(dataclasses.asdict(self.__data)) + # Written aside and moved into place: opening the cache for writing truncates it + # first, and a reader landing in that window would find the file empty and treat + # the session as broken. + fd, tmp_file = tempfile.mkstemp(dir=self.__cache_dir, prefix=".tmp-") + try: + with os.fdopen(fd, "w") as f: + f.write(payload) + # ensure file is secure. + os.chmod(tmp_file, 0o600) + self.__replace(tmp_file, self.__cache_file) + except BaseException: + if os.path.exists(tmp_file): + os.unlink(tmp_file) + raise + self.__serial = self.__stat_serial() + + @staticmethod + def __replace(source: str, destination: str) -> None: + # Windows refuses the rename while another process still has the destination + # open, which a reader briefly does, so give it a moment rather than failing the + # save outright. On POSIX the rename always succeeds and the loop ends at once. + deadline = time.monotonic() + 5.0 + while True: + try: + os.replace(source, destination) + return + except PermissionError: + if time.monotonic() >= deadline: + raise + time.sleep(0.05) + + def __stat_serial(self) -> int: stat = os.stat(self.__cache_file) - self.__serial = hash((stat.st_uid, stat.st_gid, stat.st_mode, stat.st_size, stat.st_mtime)) - # ensure file is secure. - os.chmod(self.__cache_file, 0o600) + # st_ino and st_mtime_ns both move when the file is replaced, which a refresh + # that happens to produce the same number of bytes otherwise would not show. + return hash((stat.st_ino, stat.st_uid, stat.st_gid, stat.st_mode, stat.st_size, stat.st_mtime_ns)) def __ensure_cache_dir(self) -> None: if not os.path.exists(self.__cache_dir): diff --git a/mdrsclient/commands/ls.py b/mdrsclient/commands/ls.py index a3ad62c..78e91ae 100644 --- a/mdrsclient/commands/ls.py +++ b/mdrsclient/commands/ls.py @@ -7,6 +7,7 @@ from pydantic.dataclasses import dataclass from mdrsclient.api import FilesApi, FoldersApi from mdrsclient.client import MdrsClient from mdrsclient.commands.base import BaseCommand +from mdrsclient.config import build_download_url from mdrsclient.exceptions import UnauthorizedException from mdrsclient.models import File, Folder, FolderSimple, Laboratory @@ -211,7 +212,7 @@ class LsCommand(BaseCommand): # "thumbnail": file.thumbnail, "description": file.description, "metadata": file.metadata, - "download_url": f"{context.client.connection.url}/{file.download_url}", + "download_url": build_download_url(context.client.connection.url, file.download_url), "created_at": file.created_at, "updated_at": file.updated_at, } diff --git a/mdrsclient/config.py b/mdrsclient/config.py index d5ed9e9..3ea86c8 100644 --- a/mdrsclient/config.py +++ b/mdrsclient/config.py @@ -10,6 +10,31 @@ from mdrsclient.settings import CONFIG_DIRNAME from mdrsclient.utils import FileLock +def normalize_url(url: str) -> str: + """ + Check a remote URL and put it in the one form every client agrees on. + + `simple_host` is what lets a bare hostname through, so a development server on + `localhost` is as acceptable as a deployment behind a domain name. The trailing + slash goes because the URL is joined with a path that brings its own. + """ + if not validators.url(url, simple_host=True, validate_scheme=lambda scheme: scheme in ("http", "https")): + raise IllegalArgumentException("malformed URI sequence") + return url.rstrip("/") + + +def build_download_url(base_url: str | None, path: str) -> str: + """ + Join a remote base URL with a path the API returned. + + The API answers with a relative path and no leading separator, and a configuration + written before the URL was normalised may still carry a trailing one. + """ + if path.startswith(("http://", "https://")): + return path + return f"{(base_url or '').rstrip('/')}/{path.lstrip('/')}" + + @runtime_checkable class ConfigInterface(Protocol): remote: str @@ -42,8 +67,7 @@ class InMemoryConfig(ConfigInterface): @url.setter def url(self, url: str) -> None: - if not validators.url(url): - raise IllegalArgumentException("malformed URI sequence") + url = normalize_url(url) with self.__lock: self.__configs[self.remote] = url @@ -91,8 +115,7 @@ class ConfigFile(ConfigInterface): @url.setter def url(self, url: str) -> None: - if not validators.url(url): - raise IllegalArgumentException("malformed URI sequence") + url = normalize_url(url) self.__load() if self.__config.has_section(self.remote): self.__config.remove_section(self.remote) diff --git a/mdrsclient/connection.py b/mdrsclient/connection.py index ea411bb..4f3eac9 100644 --- a/mdrsclient/connection.py +++ b/mdrsclient/connection.py @@ -1,5 +1,6 @@ import platform import threading +from contextlib import AbstractContextManager from typing import TypedDict from requests import Response, Session @@ -23,6 +24,7 @@ class _KwArgsMDRSConnectionPost(TypedDict, total=False): params: dict[str, str | int] data: dict[str, str | int] | MultipartEncoder headers: dict[str, str] + timeout: float | tuple[float, float] class _KwArgsMDRSConnectionPut(TypedDict, total=False): @@ -61,6 +63,14 @@ class MDRSConnection: def delete(self, url: str, **kwargs: Unpack[_KwArgsMDRSConnectionDelete]) -> Response: return self.session.delete(self.__build_url(url), **kwargs) + def cache_lock(self) -> AbstractContextManager[None]: + """Hold exclusive access to the login cache across every process using it.""" + return self.__cache.lock() + + def reload_cache(self) -> None: + """Re-read the login cache, discarding anything held from an earlier read.""" + self.__cache.reload() + def logout(self) -> None: del self.__cache.user del self.__cache.token @@ -96,7 +106,9 @@ class MDRSConnection: return path if self.url == "": raise MissingConfigurationException("remote host is not configured") - return f"{self.url}/{path}" + # The path brings its own separator, and a configuration written before the URL + # was normalised may still carry a trailing slash of its own. + return f"{self.url.rstrip('/')}/{path}" def __prepare_headers(self) -> None: self.session.headers.update( diff --git a/mdrsclient/utils.py b/mdrsclient/utils.py index 9b088b1..37a1359 100644 --- a/mdrsclient/utils.py +++ b/mdrsclient/utils.py @@ -1,5 +1,6 @@ import os -from typing import IO, Any +import time +from typing import IO, Any, Final from urllib.parse import parse_qs, urlparse if os.name == "nt": @@ -9,10 +10,23 @@ elif os.name == "posix": class FileLock: + # Long enough to outlast a token refresh, which is what the lock is held across. + WAIT_SECONDS: Final[float] = 60.0 + @staticmethod def lock(file: IO[Any]) -> None: if os.name == "nt": - msvcrt.locking(file.fileno(), msvcrt.LK_LOCK, 1) + # msvcrt.LK_LOCK gives up after ten one-second attempts, which is shorter + # than the refresh it now has to wait for, so do the waiting here instead. + deadline = time.monotonic() + FileLock.WAIT_SECONDS + while True: + try: + msvcrt.locking(file.fileno(), msvcrt.LK_NBLCK, 1) + return + except OSError: + if time.monotonic() >= deadline: + raise + time.sleep(0.1) elif os.name == "posix": fcntl.flock(file.fileno(), fcntl.LOCK_EX) diff --git a/tests/test_token_cache.py b/tests/test_token_cache.py new file mode 100644 index 0000000..f9d12e0 --- /dev/null +++ b/tests/test_token_cache.py @@ -0,0 +1,116 @@ +import tempfile +import time +import unittest +from unittest.mock import patch + +import jwt + +from mdrsclient.api.utils import token_check +from mdrsclient.cache import CacheFile +from mdrsclient.config import normalize_url +from mdrsclient.connection import MDRSConnection +from mdrsclient.exceptions import IllegalArgumentException +from mdrsclient.models import Token + +REMOTE = "unittest" + + +def make_token(access_offset: int, refresh_offset: int, label: str) -> Token: + now = int(time.time()) + + def encode(token_type: str, offset: int) -> str: + return jwt.encode( + { + "token_type": token_type, + "exp": now + offset, + "iat": now, + "jti": f"{label}-{token_type}", + "user_id": 1, + }, + "unittest-signing-key-not-verified-anywhere", + ) + + return Token(access=encode("access", access_offset), refresh=encode("refresh", refresh_offset)) + + +class TestTokenCache(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + patcher = patch("mdrsclient.cache.CONFIG_DIRNAME", self.tmp.name) + patcher.start() + self.addCleanup(patcher.stop) + self.addCleanup(self.tmp.cleanup) + + def test_reload_picks_up_a_write_from_another_holder(self) -> None: + """The cache is shared, so a value read earlier can already be out of date.""" + reader = CacheFile(REMOTE) + writer = CacheFile(REMOTE) + + first = make_token(3600, 86400, "first") + writer.token = first + self.assertEqual(reader.token, first) + + second = make_token(3600, 86400, "second") + writer.token = second + + reader.reload() + self.assertEqual(reader.token, second) + + def test_token_check_uses_the_token_another_holder_just_wrote(self) -> None: + """A rotating provider drops the token it replaces, so the refresh must not + be sent again once someone else has already made the round trip.""" + connection = MDRSConnection(REMOTE, "http://localhost:8000/api/") + connection.token = make_token(-60, 86400, "stale") + + # Another process refreshes while this one is between requests. + other = CacheFile(REMOTE) + rotated = make_token(3600, 86400, "rotated") + other.token = rotated + + with patch("mdrsclient.api.utils.UsersApi") as users_api: + token_check(connection) + + users_api.assert_not_called() + self.assertEqual(connection.token, rotated) + + def test_token_check_refreshes_when_nothing_else_has(self) -> None: + connection = MDRSConnection(REMOTE, "http://localhost:8000/api/") + connection.token = make_token(-60, 86400, "stale") + rotated = make_token(3600, 86400, "rotated") + + with patch("mdrsclient.api.utils.UsersApi") as users_api: + users_api.return_value.tokenRefresh.return_value = rotated + token_check(connection) + + users_api.return_value.tokenRefresh.assert_called_once() + self.assertEqual(connection.token, rotated) + + def test_token_check_logs_out_once_the_refresh_token_expires(self) -> None: + connection = MDRSConnection(REMOTE, "http://localhost:8000/api/") + connection.token = make_token(-3600, -60, "dead") + + with patch("mdrsclient.api.utils.UsersApi") as users_api: + token_check(connection) + + users_api.assert_not_called() + self.assertIsNone(connection.token) + + +class TestUrlNormalization(unittest.TestCase): + """Both clients share config.ini, so they have to agree on what a remote URL is.""" + + def test_trailing_slash_is_dropped(self) -> None: + self.assertEqual(normalize_url("http://127.0.0.1:8000/api/"), "http://127.0.0.1:8000/api") + self.assertEqual(normalize_url("https://neurodata.riken.jp/api/"), "https://neurodata.riken.jp/api") + + def test_bare_hostname_is_accepted(self) -> None: + self.assertEqual(normalize_url("http://localhost:8000/api"), "http://localhost:8000/api") + + def test_only_http_schemes_are_accepted(self) -> None: + for url in ("ftp://x.example.com/", "file:///etc/passwd", "not-a-url", "http://"): + with self.subTest(url=url), self.assertRaises(IllegalArgumentException): + normalize_url(url) + + +if __name__ == "__main__": + unittest.main()