mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-01 10:22:08 +00:00
i18n: complete Slovak PBS dialogs, status and errors
This commit is contained in:
VAIO73
@@ -8,6 +8,10 @@ from jwt_middleware import require_auth, require_admin_scope
|
||||
pbs_bp = Blueprint("pbs", __name__)
|
||||
|
||||
|
||||
def pbs_error(exc, status):
|
||||
return jsonify(error=str(exc), **exc.problem()), status
|
||||
|
||||
|
||||
@pbs_bp.after_request
|
||||
def no_cache(response):
|
||||
response.headers["Cache-Control"] = "no-store"
|
||||
@@ -27,7 +31,7 @@ def servers():
|
||||
servers = pbs.public_servers()
|
||||
return jsonify(servers=servers, local_available=pbs.local_available(), detected=_detected(servers))
|
||||
except pbs.PBSError as exc:
|
||||
return jsonify(error=str(exc)), 500
|
||||
return pbs_error(exc, 500)
|
||||
|
||||
|
||||
def _detected(servers):
|
||||
@@ -57,42 +61,42 @@ def _protected_configuration():
|
||||
@require_admin_scope
|
||||
def save_server():
|
||||
if not _protected_configuration():
|
||||
return jsonify(error="Enable local authentication before saving remote PBS credentials."), 409
|
||||
return jsonify(error="Enable local authentication before saving remote PBS credentials.", code="pbs_auth_required"), 409
|
||||
try:
|
||||
identifier = pbs.save_server(request.get_json(silent=True))
|
||||
return jsonify(success=True, id=identifier, servers=pbs.public_servers())
|
||||
except pbs.PBSError as exc:
|
||||
return jsonify(error=str(exc)), 400
|
||||
return pbs_error(exc, 400)
|
||||
except OSError:
|
||||
return jsonify(error="Could not save PBS connection."), 500
|
||||
return jsonify(error="Could not save PBS connection.", code="pbs_save_failed"), 500
|
||||
|
||||
|
||||
@pbs_bp.route("/api/pbs/servers/login", methods=["POST"])
|
||||
@require_admin_scope
|
||||
def connect_with_password():
|
||||
if not _protected_configuration():
|
||||
return jsonify(error="Enable local authentication before saving remote PBS credentials."), 409
|
||||
return jsonify(error="Enable local authentication before saving remote PBS credentials.", code="pbs_auth_required"), 409
|
||||
try:
|
||||
identifier = pbs.connect_with_password(request.get_json(silent=True))
|
||||
return jsonify(success=True, id=identifier, servers=pbs.public_servers())
|
||||
except pbs.PBSError as exc:
|
||||
return jsonify(error=str(exc)), 400
|
||||
return pbs_error(exc, 400)
|
||||
except OSError:
|
||||
return jsonify(error="Could not save PBS connection."), 500
|
||||
return jsonify(error="Could not save PBS connection.", code="pbs_save_failed"), 500
|
||||
|
||||
|
||||
@pbs_bp.route("/api/pbs/servers/<identifier>", methods=["DELETE"])
|
||||
@require_admin_scope
|
||||
def delete_server(identifier):
|
||||
if not _protected_configuration():
|
||||
return jsonify(error="Enable local authentication before changing PBS connections."), 409
|
||||
return jsonify(error="Enable local authentication before changing PBS connections.", code="pbs_auth_required"), 409
|
||||
try:
|
||||
pbs.delete_server(identifier)
|
||||
return jsonify(success=True)
|
||||
except pbs.PBSError as exc:
|
||||
return jsonify(error=str(exc)), 400
|
||||
return pbs_error(exc, 400)
|
||||
except OSError:
|
||||
return jsonify(error="Could not remove PBS connection."), 500
|
||||
return jsonify(error="Could not remove PBS connection.", code="pbs_remove_failed"), 500
|
||||
|
||||
|
||||
@pbs_bp.route("/api/pbs/status")
|
||||
@@ -101,4 +105,4 @@ def status():
|
||||
try:
|
||||
return jsonify(pbs.get_status(request.args.get("server", "local")))
|
||||
except pbs.PBSError as exc:
|
||||
return jsonify(error=str(exc)), 502
|
||||
return pbs_error(exc, 502)
|
||||
|
||||
@@ -39,7 +39,13 @@ MAX_RESPONSE = 4 * 1024 * 1024
|
||||
|
||||
|
||||
class PBSError(ValueError):
|
||||
pass
|
||||
def __init__(self, message, *, code="pbs_unknown", params=None):
|
||||
super().__init__(message)
|
||||
self.code = code
|
||||
self.params = params
|
||||
|
||||
def problem(self):
|
||||
return {"code": self.code, **({"params": self.params} if self.params else {})}
|
||||
|
||||
|
||||
def get_platform():
|
||||
@@ -87,7 +93,7 @@ def _load_servers():
|
||||
except FileNotFoundError:
|
||||
return []
|
||||
except (OSError, ValueError) as exc:
|
||||
raise PBSError("Could not read the PBS server configuration.") from exc
|
||||
raise PBSError("Could not read the PBS server configuration.", code="pbs_config_read") from exc
|
||||
|
||||
|
||||
def _save_servers(servers):
|
||||
@@ -113,11 +119,11 @@ def public_servers():
|
||||
|
||||
def _validate_server(data, previous=None, fields=("name", "url", "token_id", "token_secret", "fingerprint")):
|
||||
if not isinstance(data, dict):
|
||||
raise PBSError("Invalid PBS server configuration.")
|
||||
raise PBSError("Invalid PBS server configuration.", code="pbs_invalid_config")
|
||||
previous = previous or {}
|
||||
row = {field: data.get(field, previous.get(field, "")) for field in fields}
|
||||
if any(not isinstance(value, str) or len(value) > 4096 for value in row.values()):
|
||||
raise PBSError("Invalid PBS connection field.")
|
||||
raise PBSError("Invalid PBS connection field.", code="pbs_invalid_field")
|
||||
row["name"] = row["name"].strip()
|
||||
row["url"] = row["url"].strip().rstrip("/")
|
||||
try:
|
||||
@@ -127,26 +133,26 @@ def _validate_server(data, previous=None, fields=("name", "url", "token_id", "to
|
||||
or parts.query or parts.fragment):
|
||||
raise ValueError()
|
||||
except ValueError as exc:
|
||||
raise PBSError("Use an HTTPS PBS URL without embedded credentials, query or fragment.") from exc
|
||||
raise PBSError("Use an HTTPS PBS URL without embedded credentials, query or fragment.", code="pbs_invalid_url") from exc
|
||||
if not row["name"] or len(row["name"]) > 128:
|
||||
raise PBSError("A server name of at most 128 characters is required.")
|
||||
raise PBSError("A server name of at most 128 characters is required.", code="pbs_name_required")
|
||||
if "token_id" not in row:
|
||||
return _validated_fingerprint(row)
|
||||
row["token_id"] = row["token_id"].strip()
|
||||
if not re.fullmatch(r"[^\s:!]+@[^\s:!]+![^\s:!]+", row["token_id"]):
|
||||
raise PBSError("Enter a PBS API token ID in user@realm!token format.")
|
||||
raise PBSError("Enter a PBS API token ID in user@realm!token format.", code="pbs_token_id_format")
|
||||
if (not row["token_secret"] or not row["token_secret"].isascii()
|
||||
or any(char.isspace() or ord(char) < 32 for char in row["token_secret"])):
|
||||
raise PBSError("A valid API token secret is required.")
|
||||
raise PBSError("A valid API token secret is required.", code="pbs_token_secret")
|
||||
if not row["token_id"].isascii():
|
||||
raise PBSError("Invalid API token ID.")
|
||||
raise PBSError("Invalid API token ID.", code="pbs_invalid_token_id")
|
||||
return _validated_fingerprint(row)
|
||||
|
||||
|
||||
def _validated_fingerprint(row):
|
||||
fingerprint = row["fingerprint"].strip().lower().replace(":", "")
|
||||
if fingerprint and not re.fullmatch("[a-f0-9]{64}", fingerprint):
|
||||
raise PBSError("The certificate fingerprint must be SHA256 (64 hex characters).")
|
||||
raise PBSError("The certificate fingerprint must be SHA256 (64 hex characters).", code="pbs_invalid_fingerprint")
|
||||
row["fingerprint"] = fingerprint
|
||||
return row
|
||||
|
||||
@@ -158,7 +164,8 @@ def _remote_get(server, endpoint):
|
||||
|
||||
|
||||
def _remote_request(server, method, endpoint, headers, form=None,
|
||||
rejected="PBS rejected the API token or its permissions. Check both the user and token ACLs."):
|
||||
rejected="PBS rejected the API token or its permissions. Check both the user and token ACLs.",
|
||||
*, rejected_code="pbs_token_rejected"):
|
||||
"""One request to PBS; credentials leave only after the certificate is checked."""
|
||||
parts = urlsplit(server["url"])
|
||||
fingerprint = server.get("fingerprint")
|
||||
@@ -175,7 +182,7 @@ def _remote_request(server, method, endpoint, headers, form=None,
|
||||
if fingerprint:
|
||||
actual = hashlib.sha256(connection.sock.getpeercert(binary_form=True)).hexdigest()
|
||||
if not hmac.compare_digest(actual, fingerprint):
|
||||
raise PBSError("PBS certificate fingerprint does not match. No credentials were sent.")
|
||||
raise PBSError("PBS certificate fingerprint does not match. No credentials were sent.", code="pbs_fingerprint_mismatch")
|
||||
path = parts.path.rstrip("/") + "/api2/json" + endpoint
|
||||
body = urlencode(form) if form is not None else None
|
||||
request_headers = {"Accept": "application/json", **headers}
|
||||
@@ -185,17 +192,17 @@ def _remote_request(server, method, endpoint, headers, form=None,
|
||||
response = connection.getresponse()
|
||||
content = response.read(MAX_RESPONSE + 1)
|
||||
if response.status in (401, 403):
|
||||
raise PBSError(rejected)
|
||||
raise PBSError(rejected, code=rejected_code)
|
||||
if response.status != 200 or len(content) > MAX_RESPONSE:
|
||||
raise PBSError("PBS returned an invalid response. Redirects are not followed.")
|
||||
raise PBSError("PBS returned an invalid response. Redirects are not followed.", code="pbs_invalid_response")
|
||||
document = json.loads(content)
|
||||
if not isinstance(document, dict) or "data" not in document:
|
||||
raise PBSError("PBS returned an invalid JSON document.")
|
||||
raise PBSError("PBS returned an invalid JSON document.", code="pbs_invalid_json")
|
||||
return document["data"]
|
||||
except PBSError:
|
||||
raise
|
||||
except (OSError, http.client.HTTPException, ValueError) as exc:
|
||||
raise PBSError("Could not connect to PBS over verified HTTPS. Check the address, certificate and service.") from exc
|
||||
raise PBSError("Could not connect to PBS over verified HTTPS. Check the address, certificate and service.", code="pbs_connect_failed") from exc
|
||||
finally:
|
||||
connection.close()
|
||||
|
||||
@@ -203,19 +210,19 @@ def _remote_request(server, method, endpoint, headers, form=None,
|
||||
def save_server(data):
|
||||
identifier = data.get("id") if isinstance(data, dict) else None
|
||||
if identifier is not None and not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", str(identifier)):
|
||||
raise PBSError("Invalid PBS server ID.")
|
||||
raise PBSError("Invalid PBS server ID.", code="pbs_invalid_id")
|
||||
with _lock:
|
||||
previous = next((row for row in _load_servers() if row["id"] == identifier), None)
|
||||
row = _validate_server(data, previous)
|
||||
# /version may be public; validate credentials on the protected status API.
|
||||
usage = _remote_get(row, "/status/datastore-usage")
|
||||
if not isinstance(usage, list):
|
||||
raise PBSError("PBS returned an invalid datastore list.")
|
||||
raise PBSError("PBS returned an invalid datastore list.", code="pbs_invalid_datastores")
|
||||
row.update(id=identifier or "pbs-" + secrets.token_hex(6), revision=secrets.token_hex(16))
|
||||
with _lock:
|
||||
servers = _load_servers()
|
||||
if previous is None and len(servers) >= 32:
|
||||
raise PBSError("At most 32 PBS connections may be configured.")
|
||||
raise PBSError("At most 32 PBS connections may be configured.", code="pbs_connection_limit", params={"count": 32})
|
||||
servers = [existing for existing in servers if existing["id"] != row["id"]]
|
||||
_save_servers([*servers, row])
|
||||
return row["id"]
|
||||
@@ -228,27 +235,27 @@ def connect_with_password(data):
|
||||
username = data.get("username") if isinstance(data, dict) else None
|
||||
password = data.get("password") if isinstance(data, dict) else None
|
||||
if not isinstance(username, str) or not re.fullmatch(r"[^\s:!@]+@[^\s:!@]+", username.strip()):
|
||||
raise PBSError("Enter the PBS user as user@realm, for example root@pam.")
|
||||
raise PBSError("Enter the PBS user as user@realm, for example root@pam.", code="pbs_username")
|
||||
if not isinstance(password, str) or not password or len(password) > 4096:
|
||||
raise PBSError("Enter the password of the PBS user.")
|
||||
raise PBSError("Enter the password of the PBS user.", code="pbs_password")
|
||||
ticket = _remote_request(endpoint, "POST", "/access/ticket", {},
|
||||
{"username": username.strip(), "password": password},
|
||||
rejected="PBS rejected the user name or password.")
|
||||
rejected="PBS rejected the user name or password.", rejected_code="pbs_login_rejected")
|
||||
if not isinstance(ticket, dict) or not ticket.get("ticket") or not ticket.get("CSRFPreventionToken"):
|
||||
raise PBSError("PBS returned an invalid sign-in response.")
|
||||
raise PBSError("PBS returned an invalid sign-in response.", code="pbs_invalid_login")
|
||||
session = {"Cookie": "PBSAuthCookie=" + quote(ticket["ticket"], safe=""),
|
||||
"CSRFPreventionToken": ticket["CSRFPreventionToken"]}
|
||||
user = str(ticket.get("username") or username.strip())
|
||||
name = "proxmenux-" + secrets.token_hex(4)
|
||||
token_path = f"/access/users/{quote(user, safe='')}/token/{name}"
|
||||
created = _remote_request(endpoint, "POST", token_path, session, {"comment": "ProxMenux Monitor, read-only"},
|
||||
rejected="This PBS user cannot create API tokens.")
|
||||
rejected="This PBS user cannot create API tokens.", rejected_code="pbs_token_create_denied")
|
||||
if not isinstance(created, dict) or not created.get("tokenid") or not created.get("value"):
|
||||
raise PBSError("PBS returned an invalid API token.")
|
||||
raise PBSError("PBS returned an invalid API token.", code="pbs_invalid_token")
|
||||
try:
|
||||
_remote_request(endpoint, "PUT", "/access/acl", session,
|
||||
{"path": "/", "role": "Audit", "auth-id": created["tokenid"], "propagate": "1"},
|
||||
rejected="This PBS user cannot grant the Audit role. Sign in with root@pam or an administrator.")
|
||||
rejected="This PBS user cannot grant the Audit role. Sign in with root@pam or an administrator.", rejected_code="pbs_audit_role_denied")
|
||||
return save_server({**endpoint, "token_id": created["tokenid"], "token_secret": created["value"]})
|
||||
except PBSError:
|
||||
try:
|
||||
@@ -263,7 +270,7 @@ def delete_server(identifier):
|
||||
servers = _load_servers()
|
||||
remaining = [row for row in servers if row["id"] != identifier]
|
||||
if len(remaining) == len(servers):
|
||||
raise PBSError("PBS server not found.")
|
||||
raise PBSError("PBS server not found.", code="pbs_not_found")
|
||||
_save_servers(remaining)
|
||||
|
||||
|
||||
@@ -276,7 +283,7 @@ def _cli(*arguments):
|
||||
raise ValueError()
|
||||
return data
|
||||
except (OSError, subprocess.SubprocessError, ValueError) as exc:
|
||||
raise PBSError("Could not read PBS " + " ".join(arguments[:2]) + ". Check its service and permissions.") from exc
|
||||
raise PBSError("Could not read PBS " + " ".join(arguments[:2]) + ". Check its service and permissions.", code="pbs_read_failed") from exc
|
||||
|
||||
|
||||
def _local_api(path):
|
||||
@@ -290,7 +297,7 @@ def _local_api(path):
|
||||
raise ValueError()
|
||||
return data
|
||||
except (OSError, subprocess.SubprocessError, ValueError) as exc:
|
||||
raise PBSError("Could not read PBS " + path + ". Check its service and permissions.") from exc
|
||||
raise PBSError("Could not read PBS " + path + ". Check its service and permissions.", code="pbs_read_failed") from exc
|
||||
|
||||
|
||||
def _local_datastores():
|
||||
@@ -305,12 +312,14 @@ def _local_datastores():
|
||||
inactive = store.get("backing-device") and not (path and os.path.ismount(path))
|
||||
if not path or inactive or store.get("maintenance-mode"):
|
||||
row["error"] = "Datastore is offline, in maintenance or has no mounted backing device."
|
||||
row["error_code"] = "pbs_datastore_offline"
|
||||
else:
|
||||
try:
|
||||
capacity = shutil.disk_usage(path)
|
||||
row.update(total=capacity.total, used=capacity.used, avail=capacity.free)
|
||||
except OSError:
|
||||
row["error"] = "Datastore filesystem is not available."
|
||||
row["error_code"] = "pbs_datastore_unavailable"
|
||||
result.append(row)
|
||||
return result
|
||||
|
||||
@@ -331,10 +340,10 @@ def get_status(identifier):
|
||||
cached = _status_cache.get(identifier)
|
||||
if cached and cached[0] > time.monotonic():
|
||||
return cached[1]
|
||||
errors = {}
|
||||
errors, error_details = {}, {}
|
||||
if identifier == "local":
|
||||
if not local_available():
|
||||
raise PBSError("The local host is not a PBS server.")
|
||||
raise PBSError("The local host is not a PBS server.", code="pbs_local_unavailable")
|
||||
name, version = get_platform().get("hostname") or socket.gethostname(), _local_version()
|
||||
services = _local_services()
|
||||
operations = {"datastores": _local_datastores,
|
||||
@@ -346,7 +355,7 @@ def get_status(identifier):
|
||||
with _lock:
|
||||
server = next((row for row in _load_servers() if row["id"] == identifier), None)
|
||||
if not server:
|
||||
raise PBSError("PBS server not found.")
|
||||
raise PBSError("PBS server not found.", code="pbs_not_found")
|
||||
metadata = _remote_get(server, "/version")
|
||||
name, version = server["name"], metadata.get("version") if isinstance(metadata, dict) else None
|
||||
services = {}
|
||||
@@ -354,18 +363,19 @@ def get_status(identifier):
|
||||
"sync": "/admin/sync", "verify": "/admin/verify", "prune": "/admin/prune"}
|
||||
operations = {key: (lambda endpoint=path: _remote_get(server, endpoint)) for key, path in endpoints.items()}
|
||||
result = {"server": identifier, "name": name, "version": version, "services": services,
|
||||
"fetched_at": int(time.time()), "errors": errors}
|
||||
"fetched_at": int(time.time()), "errors": errors, "error_details": error_details}
|
||||
with ThreadPoolExecutor(max_workers=5) as executor:
|
||||
futures = {key: executor.submit(operation) for key, operation in operations.items()}
|
||||
for key, future in futures.items():
|
||||
try:
|
||||
data = future.result()
|
||||
if not isinstance(data, list) or not all(isinstance(row, dict) for row in data):
|
||||
raise PBSError("PBS returned an invalid list.")
|
||||
raise PBSError("PBS returned an invalid list.", code="pbs_invalid_list")
|
||||
result[key] = [{**row, "name": row.get("store", row.get("name"))} for row in data] if key == "datastores" else data
|
||||
except PBSError as exc:
|
||||
result[key] = None # Unknown/forbidden is never presented as zero.
|
||||
errors[key] = str(exc)
|
||||
error_details[key] = exc.problem()
|
||||
with _lock:
|
||||
if len(_status_cache) >= 33:
|
||||
_status_cache.clear()
|
||||
|
||||
@@ -205,7 +205,7 @@ class PBSManagerTests(unittest.TestCase):
|
||||
|
||||
def test_password_sign_in_keeps_only_a_read_only_token(self):
|
||||
calls = []
|
||||
def request(server, method, endpoint, headers, form=None, rejected=""):
|
||||
def request(server, method, endpoint, headers, form=None, rejected="", **kwargs):
|
||||
calls.append((method, endpoint, form))
|
||||
if endpoint == "/access/ticket":
|
||||
return {"ticket": "PBS:root@pam:ticket", "CSRFPreventionToken": "csrf", "username": "root@pam"}
|
||||
@@ -227,7 +227,7 @@ class PBSManagerTests(unittest.TestCase):
|
||||
|
||||
def test_password_sign_in_removes_its_token_when_the_role_cannot_be_granted(self):
|
||||
calls = []
|
||||
def request(server, method, endpoint, headers, form=None, rejected=""):
|
||||
def request(server, method, endpoint, headers, form=None, rejected="", **kwargs):
|
||||
calls.append((method, endpoint))
|
||||
if endpoint == "/access/ticket":
|
||||
return {"ticket": "t", "CSRFPreventionToken": "c"}
|
||||
@@ -351,15 +351,17 @@ class TLSTests(unittest.TestCase):
|
||||
|
||||
def test_bad_pin_and_untrusted_ca_send_no_http_credentials(self):
|
||||
for fingerprint in ('0' * 64, ''):
|
||||
with self.subTest(fingerprint=fingerprint), self.assertRaises(pbs.PBSError):
|
||||
with self.subTest(fingerprint=fingerprint), self.assertRaises(pbs.PBSError) as caught:
|
||||
pbs._remote_get({**self.connection, "fingerprint": fingerprint}, '/status/datastore-usage')
|
||||
self.assertEqual(caught.exception.code, "pbs_fingerprint_mismatch" if fingerprint else "pbs_connect_failed")
|
||||
self.assertEqual(self.requests, [])
|
||||
|
||||
def test_bad_credentials_and_redirects_are_rejected(self):
|
||||
for status in (401, 403, 302):
|
||||
type(self).response_status = status
|
||||
with self.subTest(status=status), self.assertRaises(pbs.PBSError):
|
||||
with self.subTest(status=status), self.assertRaises(pbs.PBSError) as caught:
|
||||
pbs._remote_get(self.connection, '/status/datastore-usage')
|
||||
self.assertEqual(caught.exception.code, "pbs_token_rejected" if status in (401, 403) else "pbs_invalid_response")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
@@ -0,0 +1,196 @@
|
||||
"""PBS localization metadata must preserve routes, security and persistence."""
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import tempfile
|
||||
import types
|
||||
import unittest
|
||||
from unittest import mock
|
||||
from flask import Flask
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[3]
|
||||
sys.path.insert(0, str(ROOT / "AppImage/scripts"))
|
||||
import auth_manager
|
||||
import pbs_manager as pbs
|
||||
REAL_REMOTE_REQUEST = pbs._remote_request
|
||||
|
||||
|
||||
class PBSI18nTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
temp = tempfile.TemporaryDirectory(); self.addCleanup(temp.cleanup)
|
||||
self.root = Path(temp.name)
|
||||
for patch in [mock.patch.multiple(auth_manager, CONFIG_DIR=self.root, AUTH_CONFIG_FILE=self.root / "auth.json"),
|
||||
mock.patch.object(pbs, "_remote_get", return_value=[]),
|
||||
mock.patch.object(pbs, "_remote_request"),
|
||||
mock.patch.object(pbs, "local_available", return_value=False)]:
|
||||
patch.start(); self.addCleanup(patch.stop)
|
||||
(self.root / "auth.json").write_text(json.dumps({"enabled": True, "password_hash": "fixture"}))
|
||||
self.server = {"name": "Fixture", "url": "https://pbs.invalid:8007", "token_id": "audit@pbs!monitor", "token_secret": "fake-fixture-secret", "fingerprint": ""}
|
||||
self.id = pbs.save_server(self.server)
|
||||
pbs._remote_get.reset_mock(); pbs._status_cache.clear()
|
||||
auth = types.SimpleNamespace(load_auth_config=auth_manager.load_auth_config, verify_token=lambda _: "fixture",
|
||||
verify_token_full=lambda token: ("fixture", "full_admin" if token == "admin" else "read_only"))
|
||||
spec = importlib.util.spec_from_file_location("fixture_pbs_jwt", ROOT / "AppImage/scripts/jwt_middleware.py")
|
||||
middleware = importlib.util.module_from_spec(spec)
|
||||
with mock.patch.dict(sys.modules, {"auth_manager": auth}): spec.loader.exec_module(middleware)
|
||||
spec = importlib.util.spec_from_file_location("fixture_pbs_routes", ROOT / "AppImage/scripts/flask_pbs_routes.py")
|
||||
routes = importlib.util.module_from_spec(spec)
|
||||
with mock.patch.dict(sys.modules, {"jwt_middleware": middleware}): spec.loader.exec_module(routes)
|
||||
app = Flask(__name__); app.register_blueprint(routes.pbs_bp); self.client = app.test_client()
|
||||
|
||||
def send(self, method="POST", path="/api/pbs/servers", data=None, role="admin"):
|
||||
return self.client.open(path, method=method, json=data, headers={"Authorization": "Bearer " + role} if role else {})
|
||||
|
||||
def reject(self, data, message, code, path="/api/pbs/servers"):
|
||||
before = (self.root / "pbs_servers.json").read_bytes()
|
||||
with mock.patch.object(pbs, "_save_servers") as save:
|
||||
response = self.send(path=path, data=data)
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertEqual(response.json, {"error": message, "code": code})
|
||||
self.assertEqual(response.headers["Cache-Control"], "no-store")
|
||||
save.assert_not_called(); pbs._remote_get.assert_not_called(); pbs._remote_request.assert_not_called()
|
||||
self.assertEqual((self.root / "pbs_servers.json").read_bytes(), before)
|
||||
|
||||
def test_invalid_input_is_localizable_before_any_remote_or_save_effect(self):
|
||||
cases = [
|
||||
(None, "Invalid PBS server configuration.", "pbs_invalid_config"),
|
||||
({**self.server, "name": []}, "Invalid PBS connection field.", "pbs_invalid_field"),
|
||||
({**self.server, "url": "http://pbs.invalid"}, "Use an HTTPS PBS URL without embedded credentials, query or fragment.", "pbs_invalid_url"),
|
||||
({**self.server, "name": ""}, "A server name of at most 128 characters is required.", "pbs_name_required"),
|
||||
({**self.server, "token_id": "bad"}, "Enter a PBS API token ID in user@realm!token format.", "pbs_token_id_format"),
|
||||
({**self.server, "token_secret": "bad\nsecret"}, "A valid API token secret is required.", "pbs_token_secret"),
|
||||
({**self.server, "token_id": "človek@pbs!monitor"}, "Invalid API token ID.", "pbs_invalid_token_id"),
|
||||
({**self.server, "fingerprint": "invalid"}, "The certificate fingerprint must be SHA256 (64 hex characters).", "pbs_invalid_fingerprint"),
|
||||
({**self.server, "id": "../../bad"}, "Invalid PBS server ID.", "pbs_invalid_id"),
|
||||
]
|
||||
for data, message, code in cases:
|
||||
with self.subTest(code=code): self.reject(data, message, code)
|
||||
for data, message, code in [
|
||||
({**self.server, "username": "missing-realm", "password": "fake"}, "Enter the PBS user as user@realm, for example root@pam.", "pbs_username"),
|
||||
({**self.server, "username": "root@pam", "password": ""}, "Enter the password of the PBS user.", "pbs_password"),
|
||||
]:
|
||||
with self.subTest(code=code): self.reject(data, message, code, "/api/pbs/servers/login")
|
||||
|
||||
def test_authentication_and_scope_guards_precede_all_mutations(self):
|
||||
before = (self.root / "pbs_servers.json").read_bytes()
|
||||
save_patch = mock.patch.object(pbs, "_save_servers")
|
||||
save = save_patch.start(); self.addCleanup(save_patch.stop)
|
||||
for path, method in [("/api/pbs/servers", "POST"), ("/api/pbs/servers/login", "POST"), ("/api/pbs/servers/" + self.id, "DELETE")]:
|
||||
for role, status in [(None, 401), ("reader", 403)]:
|
||||
self.assertEqual(self.send(method, path, self.server, role).status_code, status)
|
||||
(self.root / "auth.json").write_text(json.dumps({"enabled": False}))
|
||||
for path in ("/api/pbs/servers", "/api/pbs/servers/login"):
|
||||
response = self.send(path=path, data=self.server)
|
||||
self.assertEqual(response.status_code, 409)
|
||||
self.assertEqual(response.json, {"error": "Enable local authentication before saving remote PBS credentials.", "code": "pbs_auth_required"})
|
||||
response = self.send("DELETE", "/api/pbs/servers/" + self.id)
|
||||
self.assertEqual(response.status_code, 409)
|
||||
self.assertEqual(response.json["code"], "pbs_auth_required")
|
||||
pbs._remote_get.assert_not_called(); pbs._remote_request.assert_not_called()
|
||||
save.assert_not_called()
|
||||
self.assertEqual((self.root / "pbs_servers.json").read_bytes(), before)
|
||||
|
||||
def test_corrupt_configuration_missing_server_and_local_unavailable_keep_status(self):
|
||||
response = self.send("DELETE", "/api/pbs/servers/absent")
|
||||
self.assertEqual((response.status_code, response.json), (400, {"error": "PBS server not found.", "code": "pbs_not_found"}))
|
||||
for identifier, code in [("absent", "pbs_not_found"), ("local", "pbs_local_unavailable")]:
|
||||
response = self.send("GET", "/api/pbs/status?server=" + identifier)
|
||||
self.assertEqual(response.status_code, 502); self.assertEqual(response.json["code"], code)
|
||||
(self.root / "pbs_servers.json").write_text("invalid json")
|
||||
response = self.send("GET", "/api/pbs/servers")
|
||||
self.assertEqual((response.status_code, response.json), (500, {"error": "Could not read the PBS server configuration.", "code": "pbs_config_read"}))
|
||||
pbs._remote_get.assert_not_called(); pbs._remote_request.assert_not_called()
|
||||
|
||||
def test_save_delete_failures_leave_previous_bytes_and_do_not_leak_details(self):
|
||||
for path, method, data, code, message in [
|
||||
("/api/pbs/servers", "POST", {**self.server, "id": self.id}, "pbs_save_failed", "Could not save PBS connection."),
|
||||
("/api/pbs/servers/" + self.id, "DELETE", None, "pbs_remove_failed", "Could not remove PBS connection."),
|
||||
]:
|
||||
before = (self.root / "pbs_servers.json").read_bytes()
|
||||
with self.subTest(method=method), mock.patch.object(pbs.os, "replace", side_effect=OSError("private filesystem detail")):
|
||||
response = self.send(method, path, data)
|
||||
self.assertEqual((response.status_code, response.json), (500, {"error": message, "code": code}))
|
||||
self.assertEqual((self.root / "pbs_servers.json").read_bytes(), before)
|
||||
|
||||
def test_partial_status_preserves_legacy_error_null_and_cache_with_additive_metadata(self):
|
||||
def remote(server, endpoint):
|
||||
if endpoint == "/version": return {"version": "4.2.5"}
|
||||
if endpoint == "/admin/verify": raise pbs.PBSError("Fixture access denied", code="pbs_token_rejected")
|
||||
if endpoint == "/admin/prune": return {"invalid": "list"}
|
||||
return []
|
||||
with mock.patch.object(pbs, "_remote_get", side_effect=remote) as remote_call:
|
||||
result = self.send("GET", "/api/pbs/status?server=" + self.id)
|
||||
count = remote_call.call_count
|
||||
self.assertEqual(self.send("GET", "/api/pbs/status?server=" + self.id).json, result.json)
|
||||
self.assertEqual(remote_call.call_count, count)
|
||||
self.assertEqual(result.status_code, 200)
|
||||
self.assertIsNone(result.json["verify"]); self.assertIsNone(result.json["prune"])
|
||||
self.assertEqual(result.json["sync"], [])
|
||||
self.assertEqual(result.json["errors"]["verify"], "Fixture access denied")
|
||||
self.assertEqual(result.json["error_details"]["verify"], {"code": "pbs_token_rejected"})
|
||||
self.assertEqual(result.json["error_details"]["prune"], {"code": "pbs_invalid_list"})
|
||||
|
||||
def test_local_datastore_errors_and_command_errors_keep_legacy_messages(self):
|
||||
with mock.patch.object(pbs, "_cli", return_value=[{"name": "offline", "path": "/fixture", "backing-device": "fixture"}]), \
|
||||
mock.patch.object(pbs.os.path, "ismount", return_value=False), mock.patch.object(pbs.shutil, "disk_usage") as usage:
|
||||
store = pbs._local_datastores()[0]
|
||||
self.assertEqual(store["error"], "Datastore is offline, in maintenance or has no mounted backing device.")
|
||||
self.assertEqual(store["error_code"], "pbs_datastore_offline"); usage.assert_not_called(); self.assertIsNone(store["total"])
|
||||
with mock.patch.object(pbs.subprocess, "run", side_effect=OSError("fixture")):
|
||||
for fn, args in [(pbs._cli, ("task", "list")), (pbs._local_api, ("/admin/sync",))]:
|
||||
with self.subTest(args=args), self.assertRaises(pbs.PBSError) as caught: fn(*args)
|
||||
self.assertEqual(caught.exception.code, "pbs_read_failed")
|
||||
self.assertIn("Check its service and permissions.", str(caught.exception))
|
||||
|
||||
def test_remote_failure_codes_follow_the_actual_request_and_keep_english(self):
|
||||
cases = [
|
||||
(401, b'{}', "pbs_token_rejected", "PBS rejected the API token or its permissions. Check both the user and token ACLs."),
|
||||
(302, b'{}', "pbs_invalid_response", "PBS returned an invalid response. Redirects are not followed."),
|
||||
(200, b'{}', "pbs_invalid_json", "PBS returned an invalid JSON document."),
|
||||
(200, b'invalid-json', "pbs_connect_failed", "Could not connect to PBS over verified HTTPS. Check the address, certificate and service."),
|
||||
]
|
||||
for status, body, code, message in cases:
|
||||
connection = mock.Mock(); connection.getresponse.return_value.status = status
|
||||
connection.getresponse.return_value.read.return_value = body
|
||||
with self.subTest(code=code), mock.patch.object(pbs.http.client, "HTTPSConnection", return_value=connection), \
|
||||
self.assertRaises(pbs.PBSError) as caught:
|
||||
REAL_REMOTE_REQUEST(self.server, "GET", "/version", {})
|
||||
self.assertEqual((str(caught.exception), caught.exception.problem()), (message, {"code": code}))
|
||||
connection.close.assert_called_once()
|
||||
for code in ("pbs_login_rejected", "pbs_token_create_denied", "pbs_audit_role_denied"):
|
||||
connection = mock.Mock(); connection.getresponse.return_value.status = 403
|
||||
connection.getresponse.return_value.read.return_value = b'{}'
|
||||
with self.subTest(code=code), mock.patch.object(pbs.http.client, "HTTPSConnection", return_value=connection), self.assertRaises(pbs.PBSError) as caught:
|
||||
REAL_REMOTE_REQUEST(self.server, "POST", "/access/ticket", {}, rejected="Original context", rejected_code=code)
|
||||
self.assertEqual((str(caught.exception), caught.exception.code), ("Original context", code))
|
||||
|
||||
def test_connection_limit_and_invalid_datastore_response_do_not_save(self):
|
||||
before = (self.root / "pbs_servers.json").read_bytes()
|
||||
with mock.patch.object(pbs, "_save_servers") as save, mock.patch.object(pbs, "_load_servers", return_value=[{"id": str(i)} for i in range(32)]):
|
||||
response = self.send(data=self.server)
|
||||
self.assertEqual(response.json, {"error": "At most 32 PBS connections may be configured.", "code": "pbs_connection_limit", "params": {"count": 32}})
|
||||
self.assertEqual(response.status_code, 400); save.assert_not_called()
|
||||
pbs._remote_get.return_value = {}
|
||||
response = self.send(data=self.server)
|
||||
self.assertEqual(response.json["code"], "pbs_invalid_datastores")
|
||||
self.assertEqual((self.root / "pbs_servers.json").read_bytes(), before)
|
||||
|
||||
def test_login_failures_keep_saved_state_and_cleanup_a_created_token(self):
|
||||
before = (self.root / "pbs_servers.json").read_bytes()
|
||||
login = {"name": "Fixture", "url": "https://pbs.invalid", "username": "root@pam", "password": "fake"}
|
||||
for stage, code in [("ticket", "pbs_invalid_login"), ("token", "pbs_invalid_token"), ("acl", "pbs_audit_role_denied")]:
|
||||
calls = []
|
||||
def request(server, method, endpoint, headers, form=None, rejected="", rejected_code="pbs_token_rejected"):
|
||||
calls.append((method, endpoint))
|
||||
if endpoint == "/access/ticket": return None if stage == "ticket" else {"ticket": "fake", "CSRFPreventionToken": "fake"}
|
||||
if method == "POST": return None if stage == "token" else {"tokenid": "audit@pbs!fixture", "value": "fake"}
|
||||
if method == "PUT": raise pbs.PBSError(rejected, code=rejected_code)
|
||||
pbs._remote_request.side_effect = request
|
||||
response = self.send(path="/api/pbs/servers/login", data=login)
|
||||
self.assertEqual(response.status_code, 400); self.assertEqual(response.json["code"], code)
|
||||
self.assertEqual((self.root / "pbs_servers.json").read_bytes(), before)
|
||||
self.assertEqual(sum(method == "DELETE" for method, _ in calls), 1 if stage == "acl" else 0)
|
||||
|
||||
|
||||
if __name__ == "__main__": unittest.main()
|
||||
Reference in New Issue
Block a user