orrisroot
3081d7fbb3
Fetching a locked folder before its password is sent returns 200 with only its listing fields, without metadata, sub_folders and path, so FoldersApi.retrieve failed with a pydantic ValidationError. The DOI route fetches the folder before it can see the lock, so a password-protected DOI folder could not be opened even with its password. - Add FolderLockedException, a subclass of UnauthorizedException, and raise it from retrieve on that short record and from the lock checks. - On a DOI folder, send the password and fetch it again when the first fetch finds it locked.
272 lines
11 KiB
Python
272 lines
11 KiB
Python
import os
|
|
import re
|
|
from unicodedata import normalize
|
|
|
|
from mdrsclient.api import DoiApi, FilesApi, FoldersApi, LaboratoriesApi, UsersApi
|
|
from mdrsclient.api.utils import token_check
|
|
from mdrsclient.cache import CacheInterface
|
|
from mdrsclient.config import ConfigFile, ConfigInterface
|
|
from mdrsclient.connection import MDRSConnection
|
|
from mdrsclient.exceptions import (
|
|
FolderLockedException,
|
|
IllegalArgumentException,
|
|
MissingConfigurationException,
|
|
UnexpectedException,
|
|
)
|
|
from mdrsclient.models import File, Folder, Laboratory, Token, User
|
|
from mdrsclient.utils import page_num_from_url
|
|
|
|
|
|
class MdrsService:
|
|
config_class: type[ConfigInterface] = ConfigFile
|
|
|
|
def __init__(self, connection: MDRSConnection | None, config_class: type[ConfigInterface] | None = None):
|
|
self.__connection = connection
|
|
if config_class is not None:
|
|
self.config_class = config_class
|
|
|
|
@property
|
|
def connection(self) -> MDRSConnection:
|
|
"""
|
|
The connection every remote operation goes through.
|
|
|
|
Optional to supply, because `config` and `version` do their work without a remote
|
|
and are reached through the same client. Asking for it when none was given is a
|
|
mistake in the caller, and says so rather than failing later on `None`.
|
|
"""
|
|
if self.__connection is None:
|
|
raise MissingConfigurationException("This operation requires a remote host.")
|
|
return self.__connection
|
|
|
|
@connection.setter
|
|
def connection(self, connection: MDRSConnection | None) -> None:
|
|
# Assignable as it always was: this is a property to check for absence on read,
|
|
# not to make the connection fixed once the client is built.
|
|
self.__connection = connection
|
|
|
|
@classmethod
|
|
def create_connection(
|
|
cls, remote: str, cache: CacheInterface | None = None, config: ConfigInterface | None = None
|
|
) -> MDRSConnection:
|
|
if config is None:
|
|
config = ConfigFile(remote)
|
|
elif config.remote != remote:
|
|
raise IllegalArgumentException("Remote host parameter mismatch.")
|
|
if config.url is None:
|
|
raise MissingConfigurationException(f"Remote host `{remote}` is not found.")
|
|
return MDRSConnection(config.remote, config.url, cache=cache)
|
|
|
|
def login(self, username: str, password: str) -> tuple[Token, User]:
|
|
user_api = UsersApi(self.connection)
|
|
token = user_api.token(username, password)
|
|
self.connection.token = token
|
|
user = user_api.current()
|
|
self.connection.user = user
|
|
return token, user
|
|
|
|
def logout(self) -> None:
|
|
self.connection.logout()
|
|
|
|
def project_names(self) -> dict[int, str]:
|
|
"""
|
|
Names of the logged-in user's projects by bit_position, for labelling access levels.
|
|
|
|
Empty when nobody is logged in: the server names a project only to its members.
|
|
"""
|
|
token_check(self.connection)
|
|
if self.connection.token is None:
|
|
return {}
|
|
return UsersApi(self.connection).project_names()
|
|
|
|
def whoami(self) -> User:
|
|
user_api = UsersApi(self.connection)
|
|
return user_api.current()
|
|
|
|
def get_laboratories(self) -> list[Laboratory]:
|
|
laboratory_api = LaboratoriesApi(self.connection)
|
|
labs = laboratory_api.list()
|
|
self.connection.laboratories = labs
|
|
return list(labs)
|
|
|
|
def find_laboratory(self, name: str) -> Laboratory:
|
|
if self.connection.laboratories.empty() or (self.connection.token and self.connection.token.is_expired):
|
|
self.get_laboratories()
|
|
laboratory = self.connection.laboratories.find_by_name(name)
|
|
if laboratory is None:
|
|
raise IllegalArgumentException(f"Laboratory `{name}` not found.")
|
|
return laboratory
|
|
|
|
def find_folder(self, laboratory: Laboratory, path: str, password: str | None = None) -> Folder:
|
|
folder_api = FoldersApi(self.connection)
|
|
folders = folder_api.list(laboratory.id, normalize("NFC", path))
|
|
if len(folders) != 1:
|
|
raise UnexpectedException(f"Folder `{path}` not found.")
|
|
if folders[0].lock:
|
|
if password is None:
|
|
raise FolderLockedException(f"Folder `{path}` is locked.")
|
|
folder_api.auth(folders[0].id, password)
|
|
return folder_api.retrieve(folders[0].id)
|
|
|
|
def find_files(self, folder_id: str) -> list[File]:
|
|
files_api = FilesApi(self.connection)
|
|
page = 1
|
|
results_file = []
|
|
while page:
|
|
result = files_api.list(folder_id, page)
|
|
results_file.extend(result.results)
|
|
page = 0
|
|
if result.next:
|
|
page = page_num_from_url(result.next)
|
|
return results_file
|
|
|
|
@staticmethod
|
|
def is_doi(path_component: str) -> bool:
|
|
return path_component.startswith("10.") and "/" in path_component
|
|
|
|
@staticmethod
|
|
def doi_suffix_id(doi: str) -> str:
|
|
doi = doi.rstrip("/")
|
|
slash_pos = doi.find("/")
|
|
if slash_pos == -1:
|
|
return doi
|
|
suffix = doi[slash_pos + 1 :]
|
|
dot_pos = suffix.rfind(".")
|
|
return suffix[dot_pos + 1 :] if dot_pos != -1 else suffix
|
|
|
|
@staticmethod
|
|
def split_doi_and_subpath(doi_with_path: str) -> tuple[str, str]:
|
|
first_slash = doi_with_path.find("/")
|
|
if first_slash != -1:
|
|
after_suffix_start = first_slash + 1
|
|
after_first = doi_with_path[after_suffix_start:]
|
|
second_slash = after_first.find("/")
|
|
if second_slash != -1:
|
|
doi_end = after_suffix_start + second_slash
|
|
doi = doi_with_path[:doi_end]
|
|
subpath = doi_with_path[doi_end:]
|
|
if subpath == "/":
|
|
return (doi, "")
|
|
else:
|
|
return (doi, subpath)
|
|
else:
|
|
return (doi_with_path, "")
|
|
else:
|
|
return (doi_with_path, "")
|
|
|
|
@classmethod
|
|
def parse_remote_host(cls, path: str) -> str:
|
|
path_array = path.split(":")
|
|
remote_host = path_array[0]
|
|
if len(path_array) == 2 and path_array[1] != "" or len(path_array) > 2:
|
|
raise IllegalArgumentException("Invalid remote host")
|
|
return remote_host
|
|
|
|
@classmethod
|
|
def parse_remote_host_with_path(cls, path: str) -> tuple[str, str, str]:
|
|
path = re.sub(r"//+|/\./+|/\.$", "/", path)
|
|
if re.search(r"/\.\./|/\.\.$", path) is not None:
|
|
raise IllegalArgumentException("Path traversal found.")
|
|
path_array = path.split(":")
|
|
if len(path_array) != 2:
|
|
raise IllegalArgumentException("Invalid remote host.")
|
|
remote_host = path_array[0]
|
|
folder_array = path_array[1].split("/")
|
|
is_absolute_path = folder_array[0] == ""
|
|
if not is_absolute_path:
|
|
raise IllegalArgumentException("Must be absolute paths.")
|
|
del folder_array[0]
|
|
if len(folder_array) == 0:
|
|
laboratory = ""
|
|
folder = ""
|
|
else:
|
|
laboratory = folder_array.pop(0)
|
|
folder = "/" + "/".join(folder_array)
|
|
return (remote_host, laboratory, folder)
|
|
|
|
@classmethod
|
|
def parse_doi_remote_host(cls, path: str) -> tuple[str, str, str]:
|
|
parts = path.split(":", 1)
|
|
if len(parts) != 2:
|
|
raise IllegalArgumentException("remote_path must be in the form 'remote:10.xxxx/prefix.ID'")
|
|
remote, doi_with_path = parts
|
|
if not cls.is_doi(doi_with_path):
|
|
raise IllegalArgumentException(f"Path `{doi_with_path}` does not look like a DOI.")
|
|
doi, subpath = cls.split_doi_and_subpath(doi_with_path)
|
|
return (remote, doi, subpath)
|
|
|
|
def find_folder_by_doi(self, doi: str, password: str | None = None) -> tuple[Folder, Laboratory]:
|
|
doi_clean = doi.rstrip("/")
|
|
doi_id = self.doi_suffix_id(doi_clean)
|
|
doi_api = DoiApi(self.connection)
|
|
doi_resp = doi_api.retrieve(doi_id)
|
|
|
|
returned_doi = doi_resp.doi.rstrip("/")
|
|
if returned_doi.lower() != doi_clean.lower():
|
|
raise IllegalArgumentException(
|
|
f"DOI mismatch: requested `{doi_clean}` but server returned `{returned_doi}`."
|
|
)
|
|
|
|
folder_api = FoldersApi(self.connection)
|
|
# The folder is found by its id, so whether it is locked is known only from what
|
|
# retrieve returns: a full record, or the short one of a locked folder.
|
|
folder: Folder | None
|
|
try:
|
|
folder = folder_api.retrieve(doi_resp.folder_id)
|
|
except FolderLockedException:
|
|
folder = None
|
|
|
|
if folder is None or folder.lock:
|
|
if password is None:
|
|
raise FolderLockedException(f"Folder for DOI `{doi_clean}` is locked.")
|
|
folder_api.auth(doi_resp.folder_id, password)
|
|
if folder is None:
|
|
folder = folder_api.retrieve(doi_resp.folder_id)
|
|
|
|
lab_api = LaboratoriesApi(self.connection)
|
|
labs = lab_api.list()
|
|
lab = labs.find_by_id(folder.laboratory_id)
|
|
if lab is None:
|
|
raise UnexpectedException(f"Laboratory with id {folder.laboratory_id} not found.")
|
|
|
|
self.connection.laboratories = labs
|
|
return (folder, lab)
|
|
|
|
def resolve_folder(self, remote_path: str, password: str | None = None) -> tuple[Folder, Laboratory]:
|
|
path_component = remote_path.split(":", 1)[1] if ":" in remote_path else ""
|
|
if self.is_doi(path_component):
|
|
remote, doi, subpath = self.parse_doi_remote_host(remote_path)
|
|
doi_folder, laboratory = self.find_folder_by_doi(doi, password)
|
|
if not subpath:
|
|
return (doi_folder, laboratory)
|
|
else:
|
|
abs_path = doi_folder.path.rstrip("/") + subpath
|
|
folder = self.find_folder(laboratory, abs_path, password)
|
|
return (folder, laboratory)
|
|
else:
|
|
remote, laboratory_name, r_path = self.parse_remote_host_with_path(remote_path)
|
|
laboratory = self.find_laboratory(laboratory_name)
|
|
folder = self.find_folder(laboratory, r_path, password)
|
|
return (folder, laboratory)
|
|
|
|
def resolve_file(self, remote_path: str, password: str | None = None) -> tuple[Folder, Laboratory, str]:
|
|
path_component = remote_path.split(":", 1)[1] if ":" in remote_path else ""
|
|
if self.is_doi(path_component):
|
|
remote, doi, subpath = self.parse_doi_remote_host(remote_path)
|
|
doi_folder, laboratory = self.find_folder_by_doi(doi, password)
|
|
subpath_clean = subpath.rstrip("/")
|
|
if not subpath_clean:
|
|
raise IllegalArgumentException("DOI path must point to a file, not a folder.")
|
|
r_dirname = os.path.dirname(subpath_clean)
|
|
r_basename = os.path.basename(subpath_clean)
|
|
abs_path = doi_folder.path.rstrip("/") + r_dirname
|
|
parent_folder = self.find_folder(laboratory, abs_path, password)
|
|
return (parent_folder, laboratory, r_basename)
|
|
else:
|
|
remote, laboratory_name, r_path = self.parse_remote_host_with_path(remote_path)
|
|
r_path = r_path.rstrip("/")
|
|
r_dirname = os.path.dirname(r_path)
|
|
r_basename = os.path.basename(r_path)
|
|
laboratory = self.find_laboratory(laboratory_name)
|
|
parent_folder = self.find_folder(laboratory, r_dirname, password)
|
|
return (parent_folder, laboratory, r_basename)
|