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
This commit is contained in:
+108
-20
@@ -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):
|
||||
|
||||
Reference in New Issue
Block a user