Maxim Shulga
e6034a69a6
set_defaults() assigned os.getenv() results directly, which overwrote the argparse defaults with None whenever a variable was unset. A default container therefore ran "seaf-cli config -k delete_confirm_threshold -v None", and the same for upload_limit and download_limit: none of the three documented defaults were ever applied. SEAFILE_UID and SEAFILE_GID had a second problem. argparse type=int applies only to command line flags, never to set_defaults, so both stayed strings. The check in setup_uid compares them against pwd integers, and a string is never equal to one, so usermod ran on every start. dsc/config.py reads each variable with its documented default and rejects a malformed or negative value at startup, naming the variable, instead of passing it on. An empty variable counts as unset. Booleans are now rendered as true and false rather than Python's True and False. The daemon reads disable_verify_certificate with its boolean accessor, which does not recognise the capitalised form. I could not run the daemon to confirm the previous value's effect, since the AppImage is x86_64 only. seaf-cli is no longer given a None value for any option.
333 lines
13 KiB
Python
333 lines
13 KiB
Python
import argparse
|
|
import json
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
import time
|
|
from typing import Optional
|
|
|
|
from cached_property import cached_property_with_ttl
|
|
import requests
|
|
from requests.adapters import HTTPAdapter
|
|
from urllib3.util.retry import Retry
|
|
import seafile
|
|
|
|
from dsc import const
|
|
from dsc.errors import ConfigError, DaemonError, DaemonTimeout
|
|
from dsc.misc import config_value, create_dir, hide_secrets, user_cmd
|
|
|
|
_lg = logging.getLogger(__name__)
|
|
|
|
|
|
class SeafileClient:
|
|
def __init__(self,
|
|
host: str,
|
|
user: str,
|
|
passwd: str = None,
|
|
app_dir: str = const.DEFAULT_APP_DIR,
|
|
token: str = None):
|
|
self.user = user
|
|
self.password = passwd
|
|
self.app_dir = os.path.abspath(app_dir)
|
|
self.rpc = seafile.RpcClient(os.path.join(self.app_dir, 'seafile-data', 'seafile.sock'))
|
|
# A token supplied by the operator is used as is; otherwise one is
|
|
# fetched from the server with the password.
|
|
self.__token = token
|
|
|
|
# determine server URL (assume HTTPS unless explicitly specified)
|
|
if host.startswith('http://') or host.startswith('https://'):
|
|
self.url = host.rstrip('/')
|
|
else:
|
|
self.url = f"https://{host}"
|
|
|
|
# configure session with retry strategy
|
|
# enable urllib3 retry logging at DEBUG level (shows retry attempts)
|
|
urllib3_logger = logging.getLogger("urllib3.connectionpool")
|
|
urllib3_logger.setLevel(logging.DEBUG)
|
|
urllib3_logger.propagate = True
|
|
|
|
self.session = requests.Session()
|
|
retry_strategy = Retry(
|
|
total=30,
|
|
backoff_factor=2,
|
|
backoff_max=60,
|
|
status_forcelist=[500, 502, 503, 504],
|
|
allowed_methods=["GET", "POST"]
|
|
)
|
|
adapter = HTTPAdapter(max_retries=retry_strategy)
|
|
self.session.mount("http://", adapter)
|
|
self.session.mount("https://", adapter)
|
|
|
|
def __str__(self):
|
|
return f"SeafileClient({self.user}@{self.url})"
|
|
|
|
def __gen_cmd(self, argv: list) -> list:
|
|
return user_cmd(argv)
|
|
|
|
def __run(self, argv: list, check: bool = True, **kwargs):
|
|
"""Run a seaf-cli command and, unless told otherwise, check its exit
|
|
code. An unchecked failure here surfaces much later as a daemon that
|
|
never becomes ready, with nothing pointing at the real cause."""
|
|
proc = subprocess.run(self.__gen_cmd(argv), **kwargs)
|
|
if check and proc.returncode != 0:
|
|
raise DaemonError(
|
|
f"Command {' '.join(argv)} failed with exit code {proc.returncode}"
|
|
)
|
|
return proc
|
|
|
|
def __wait_for_daemon(self, ready: bool, timeout: float, action: str):
|
|
"""Wait until the daemon is ready (or stopped), but never forever."""
|
|
deadline = time.monotonic() + timeout
|
|
while self.daemon_ready != ready:
|
|
if time.monotonic() >= deadline:
|
|
state = "ready" if ready else "stopped"
|
|
raise DaemonTimeout(
|
|
f"Seafile daemon did not become {state} within {timeout}s"
|
|
f" while {action}"
|
|
)
|
|
time.sleep(const.DAEMON_POLL_PERIOD)
|
|
|
|
@property
|
|
def token(self):
|
|
if self.__token is None:
|
|
if not self.password:
|
|
raise ConfigError("No password or token available to authenticate")
|
|
url = f"{self.url}/api2/auth-token/"
|
|
_lg.info("Fetching token: %s", url)
|
|
r = self.session.post(url, data={"username": self.user, "password": self.password})
|
|
if r.status_code != 200:
|
|
raise RuntimeError(f"Can't get token: {r.text}")
|
|
self.__token = r.json()["token"]
|
|
return self.__token
|
|
|
|
@cached_property_with_ttl(ttl=60)
|
|
def remote_libraries(self) -> dict:
|
|
url = f"{self.url}/api2/repos/"
|
|
_lg.info("Fetching remote libraries: %s", url)
|
|
auth_header = {"Authorization": f"Token {self.token}"}
|
|
r = self.session.get(url, headers=auth_header)
|
|
if r.status_code != 200:
|
|
raise RuntimeError(r.text)
|
|
r_libs = {lib["id"]: lib["name"] for lib in r.json()}
|
|
return r_libs
|
|
|
|
@property
|
|
def config_initialized(self) -> bool:
|
|
return os.path.isdir(os.path.join(self.app_dir, ".ccnet"))
|
|
|
|
@property
|
|
def daemon_ready(self) -> bool:
|
|
cmd = ["seaf-cli", "status"]
|
|
_lg.info("Checking seafile daemon status: %s", " ".join(cmd))
|
|
proc = subprocess.run(
|
|
self.__gen_cmd(cmd),
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
return proc.returncode == 0
|
|
|
|
def init_config(self):
|
|
if self.config_initialized:
|
|
return
|
|
cmd = ["seaf-cli", "init", "-d", self.app_dir]
|
|
_lg.info("Initializing seafile config: %s", " ".join(cmd))
|
|
self.__run(cmd)
|
|
|
|
def start_daemon(self):
|
|
cmd = ["seaf-cli", "start"]
|
|
_lg.info("Starting seafile daemon: %s", " ".join(cmd))
|
|
self.__run(cmd)
|
|
_lg.info("Waiting up to %ss for seafile daemon to start",
|
|
const.DAEMON_START_TIMEOUT)
|
|
self.__wait_for_daemon(True, const.DAEMON_START_TIMEOUT, "starting")
|
|
_lg.info("Seafile daemon is ready")
|
|
|
|
def stop_daemon(self):
|
|
cmd = ["seaf-cli", "stop"]
|
|
_lg.info("Stopping seafile daemon: %s", " ".join(cmd))
|
|
# A daemon that is already gone makes "stop" fail; the state below is
|
|
# what matters, not this exit code.
|
|
self.__run(cmd, check=False)
|
|
_lg.info("Waiting up to %ss for seafile daemon to stop",
|
|
const.DAEMON_STOP_TIMEOUT)
|
|
self.__wait_for_daemon(False, const.DAEMON_STOP_TIMEOUT, "stopping")
|
|
_lg.info("Seafile daemon is stopped")
|
|
|
|
def get_library_id(self, library) -> Optional[str]:
|
|
for lib_id, lib_name in self.remote_libraries.items():
|
|
if library in (lib_id, lib_name):
|
|
return lib_id
|
|
return None
|
|
|
|
def sync_lib(self, lib_id: str, lib_dir: str) -> bool:
|
|
"""
|
|
Sync a library into an already resolved directory. The caller decides
|
|
the directory, see dsc.paths.plan_lib_dirs: the name comes from the
|
|
server and must not be turned into a path here.
|
|
|
|
Returns whether the library was accepted. One library that cannot be
|
|
synced must not stop the others, so this reports instead of raising.
|
|
"""
|
|
create_dir(lib_dir)
|
|
cmd = [
|
|
"seaf-cli",
|
|
"sync",
|
|
"-l", lib_id,
|
|
"-s", self.url,
|
|
"-d", lib_dir,
|
|
"-u", self.user,
|
|
# The token, not the password: seaf-cli ignores the password when a
|
|
# token is given, and command arguments are visible to any process
|
|
# in the container. A token can also be revoked on the server.
|
|
"-T", self.token,
|
|
]
|
|
_lg.info(
|
|
"Syncing library %s into %s: %s", lib_id, lib_dir,
|
|
" ".join(hide_secrets(cmd, self.token, self.password)),
|
|
)
|
|
proc = self.__run(cmd, check=False)
|
|
if proc.returncode != 0:
|
|
_lg.error("Failed to sync library %s into %s: seaf-cli exited with %s",
|
|
lib_id, lib_dir, proc.returncode)
|
|
return False
|
|
return True
|
|
|
|
def __print_tx_task(self, tx_task) -> str:
|
|
""" Print transfer task status """
|
|
try:
|
|
percentage = tx_task.block_done / tx_task.block_total * 100
|
|
tx_rate = tx_task.rate / 1024.0
|
|
return f" {percentage:.1f}%, {tx_rate:.1f}KB/s"
|
|
except ZeroDivisionError:
|
|
return ""
|
|
|
|
def get_status(self) -> dict:
|
|
""" Get status of all libraries """
|
|
statuses = dict()
|
|
|
|
# fetch statuses of libraries being cloned
|
|
tasks = self.rpc.get_clone_tasks()
|
|
for clone_task in tasks:
|
|
if clone_task.state == "done":
|
|
continue
|
|
|
|
elif clone_task.state == "fetch":
|
|
statuses[clone_task.repo_name] = "downloading"
|
|
tx_task = self.rpc.find_transfer_task(clone_task.repo_id)
|
|
statuses[clone_task.repo_name] += self.__print_tx_task(tx_task)
|
|
|
|
elif clone_task.state == "error":
|
|
err = self.rpc.sync_error_id_to_str(clone_task.error)
|
|
statuses[clone_task.repo_name] = f"error: {err}"
|
|
|
|
else:
|
|
statuses[clone_task.repo_name] = clone_task.state
|
|
|
|
# fetch statuses of synced libraries
|
|
repos = self.rpc.get_repo_list(-1, -1)
|
|
for repo in repos:
|
|
auto_sync_enabled = self.rpc.is_auto_sync_enabled()
|
|
if not auto_sync_enabled or not repo.auto_sync:
|
|
statuses[repo.name] = "auto sync disabled"
|
|
continue
|
|
|
|
sync_task = self.rpc.get_repo_sync_task(repo.id)
|
|
if sync_task is None:
|
|
statuses[repo.name] = "waiting for sync"
|
|
|
|
elif sync_task.state in ("uploading", "downloading"):
|
|
statuses[repo.name] = sync_task.state
|
|
tx_task = self.rpc.find_transfer_task(repo.id)
|
|
|
|
if sync_task.state == "downloading":
|
|
if tx_task.rt_state == "data":
|
|
statuses[repo.name] += " files"
|
|
elif tx_task.rt_state == "fs":
|
|
statuses[repo.name] += " file list"
|
|
|
|
statuses[repo.name] += self.__print_tx_task(tx_task)
|
|
|
|
elif sync_task.state == "error":
|
|
err = self.rpc.sync_error_id_to_str(sync_task.error)
|
|
statuses[repo.name] = f"error: {err}"
|
|
|
|
else:
|
|
statuses[repo.name] = sync_task.state
|
|
|
|
return statuses
|
|
|
|
def watch_status(self):
|
|
prev_status = dict()
|
|
max_name_len = 0
|
|
fmt = "Library {:%ds} {}" % max_name_len
|
|
while True:
|
|
time.sleep(const.STATUS_POLL_PERIOD)
|
|
cur_status = self.get_status()
|
|
for library, state in cur_status.items():
|
|
if state != prev_status.get(library):
|
|
if 30 > len(library) > max_name_len:
|
|
max_name_len = len(library)
|
|
fmt = "Library {:%ds} {}" % max_name_len
|
|
logging.info(fmt.format(library, state))
|
|
prev_status[library] = cur_status[library]
|
|
|
|
def get_local_libraries(self) -> set:
|
|
# The display output is whitespace separated, so a library name or a
|
|
# path containing a space cannot be parsed back apart. Ask for JSON.
|
|
cmd = ["seaf-cli", "list", "--json"]
|
|
_lg.info("Listing local libraries: %s", " ".join(cmd))
|
|
out = self.__run(cmd, stdout=subprocess.PIPE).stdout
|
|
|
|
try:
|
|
libraries = json.loads(out.decode())
|
|
except (UnicodeDecodeError, ValueError) as err:
|
|
raise DaemonError(
|
|
f"Cannot read the library list from seaf-cli: {err}"
|
|
) from err
|
|
|
|
# Valid JSON is not necessarily the expected shape, and an unexpected
|
|
# one must fail the same way rather than as a bare TypeError.
|
|
try:
|
|
return {lib["id"] for lib in libraries}
|
|
except (KeyError, TypeError) as err:
|
|
raise DaemonError(
|
|
f"Unexpected library list from seaf-cli: {err}"
|
|
) from err
|
|
|
|
def configure(self, args: argparse.Namespace, check_for_daemon: bool = True):
|
|
need_restart = False
|
|
# Options can be fetched or set only when daemon is running
|
|
if check_for_daemon and not self.daemon_ready:
|
|
self.start_daemon()
|
|
|
|
for key, value in args.__dict__.items():
|
|
if key not in const.AVAILABLE_SEAFCLI_OPTIONS:
|
|
continue
|
|
|
|
if value is None:
|
|
raise ConfigError(f"No value for seafile option {key}")
|
|
|
|
# check current value
|
|
cmd = ["seaf-cli", "config", "-k", key]
|
|
_lg.info("Checking seafile client option: %s", " ".join(cmd))
|
|
proc = self.__run(cmd, stdout=subprocess.PIPE)
|
|
# stdout looks like "option = value"
|
|
cur_value = proc.stdout.decode().strip()
|
|
try:
|
|
cur_value = cur_value.split(sep="=")[1].strip()
|
|
except IndexError:
|
|
cur_value = None
|
|
if cur_value == config_value(value):
|
|
continue
|
|
|
|
# set new value
|
|
cmd = ["seaf-cli", "config", "-k", key, "-v", config_value(value)]
|
|
_lg.info("Setting seafile client option: %s", " ".join(cmd))
|
|
self.__run(cmd)
|
|
need_restart = True
|
|
|
|
if need_restart:
|
|
_lg.info("Restarting seafile daemon")
|
|
self.stop_daemon()
|
|
self.start_daemon()
|