feat(selfhost): show sanitized update logs in admin UI

scripts/selfhost-update-service.py

@@ -14,6 +14,7 @@ import socketserver
 import stat
 import threading
 import traceback
+import urllib.parse
 
 spec = importlib.util.spec_from_file_location("fedeo_updater", Path(__file__).with_name("selfhost-update.py"))
 engine = importlib.util.module_from_spec(spec)
@@ -33,7 +34,8 @@ ERRORS = {
     "COMPOSE_REVIEW_REQUIRED": "Die Compose-Datei muss zuerst auf dem Host geprüft und übernommen werden.",
     "RECOVERY_REQUIRED": "Ein vorheriges Update benötigt Wiederherstellung auf dem Host.",
     "PLAN_CHANGED": "Release oder Konfiguration haben sich geändert. Erneut prüfen.",
-    "UPDATE_FAILED": "Auftrag fehlgeschlagen. Geschützte Protokolle auf dem Host prüfen.",
+    "UPDATE_FAILED": "Auftrag fehlgeschlagen.",
+    "LOGS_UNAVAILABLE": "Das Update-Protokoll kann gerade nicht gelesen werden.",
     "INTERRUPTED": "Der Auftrag wurde unterbrochen. Status auf dem Host prüfen.",
 }
 
@@ -82,6 +84,74 @@ def public_plan(plan):
     return result
 
 
+LOG_BYTES = 6000
+SECRET_KEY = r"[\w.-]*(?:password|passwd|secret|token|credential|authorization|(?:api|private|access|encryption|signing)[_-]?key|database[_-]?url|dsn|auth)[\w.-]*"
+QUOTED_VALUE = r"\"(?:\\.|[^\"\\])*\"|'[^']*'"
+SECRET_VALUE = re.compile(r"(?im)(?<![\w.-])([\"']?" + SECRET_KEY + r"[\"']?\s*[:=]\s*)(" + QUOTED_VALUE + r"|[^\r\n,}\]]+)")
+CONFIG_SECRET = re.compile(r"(?im)^\s*(?:export\s+|-\s*)?[\"']?" + SECRET_KEY + r"[\"']?\s*[:=]\s*(" + QUOTED_VALUE + r"|[^\r\n]+)")
+
+
+def read_private(path, *, tail=False):
+    """Read only regular files, never follow a log/config symlink or block on a FIFO."""
+    try:
+        fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+    except FileNotFoundError:
+        return "", False
+    with os.fdopen(fd, "rb") as stream:
+        size = os.fstat(stream.fileno()).st_size
+        if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode):
+            raise OSError("Not a regular file")
+        limit = 32768 if tail else 1024 * 1024
+        if size > limit and not tail:
+            raise OSError("Configuration too large to redact safely")
+        truncated = size > limit
+        if truncated:
+            stream.seek(-limit, os.SEEK_END)
+            stream.readline(limit)  # Never expose a suffix of a secret on a cut line.
+        return stream.read(limit).decode("utf-8", errors="replace"), truncated
+
+
+def log_redactor(directory, job):
+    # Include original values after an update changed the active configuration.
+    paths = [directory / name for name in (".env", "docker-compose.yml", "docker-compose.override.yml", "docker-compose.override.yaml")]
+    if job:
+        paths += [job / "target.env", job / "target.compose.yml"]
+        paths += [job / "backup/config" / name for name in (".env", "docker-compose.yml", "docker-compose.override.yml", "docker-compose.override.yaml")]
+    values = {value for key, value in os.environ.items() if re.fullmatch(SECRET_KEY, key, re.I) and value}
+    for path in paths:
+        content, _ = read_private(path)
+        for match in CONFIG_SECRET.finditer(content):
+            value = match[1].strip()
+            if value[:1] in ("'", '"'):
+                if value.startswith('"'):
+                    try:
+                        values.add(json.loads(value))
+                    except ValueError:
+                        pass
+                value = value[1:-1]
+            else:
+                value = value.split(" #", 1)[0].strip()
+            if value:
+                values.add(value)
+        for match in re.finditer(r"[a-z][a-z0-9+.-]*://[^\s\"']+", content, re.I):
+            password = urllib.parse.urlsplit(match[0]).password
+            if password:
+                values.update((password, urllib.parse.unquote(password)))
+    values |= {urllib.parse.quote(value, safe="") for value in values}
+
+    def redact(content):
+        content = re.sub(r"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07]*(?:\x07|\x1b\\))", "", content)
+        for value in sorted(values, key=len, reverse=True):
+            content = content.replace(value, "[REDACTED]")
+        content = re.sub(r"(?is)-----BEGIN [^-]*PRIVATE KEY-----.*?(?:-----END [^-]*PRIVATE KEY-----|\Z)", "[REDACTED]", content)
+        content = re.sub(r"(?i)([a-z][a-z0-9+.-]*://)[^\s/]*@", r"\1[REDACTED]@", content)
+        content = re.sub(r"(?i)\b(Bearer|Basic|token)\s+[A-Za-z0-9._~+/=-]+", r"\1 [REDACTED]", content)
+        content = SECRET_VALUE.sub(lambda m: m[1] + "[REDACTED]", content)
+        content = re.sub(r"(?i)(--" + SECRET_KEY + r"\s+)(" + QUOTED_VALUE + r"|\S+)", r"\1[REDACTED]", content)
+        return re.sub(r"[\x00-\x08\x0b-\x1f\x7f]", "", content)
+    return redact
+
+
 class Controller:
     def __init__(self, directory, *, updater_factory=engine.Updater):
         self.directory = Path(directory).resolve()
@@ -140,9 +210,47 @@ class Controller:
                 result["progress"] = public_state(state)
         return result
 
+    def request_logs(self, request_id):
+        record = self.read(request_id)
+        try:
+            home = self.home.parent
+            state = engine.read_json(home / "status.json", {})
+            candidates = [Path(record["job"])] if record.get("job") else []
+            if state.get("requestId") == request_id and state.get("job"):
+                candidates.append(Path(state["job"]))
+            # Older service versions did not persist the engine-job association.
+            if not candidates:
+                candidates = [path.parent for path in home.glob("*/status.json")]
+            job = None
+            for candidate in candidates:
+                if candidate.parent != home or candidate.is_symlink() or not candidate.is_dir():
+                    continue
+                content, _ = read_private(candidate / "status.json")
+                if content and json.loads(content).get("requestId") == request_id:
+                    job = candidate
+                    break
+            redact = log_redactor(self.directory, job)
+            files = [("service", self.home / (request_id + ".log"))]
+            if job:
+                files += [("events", job / "events.jsonl"), ("commands", job / "commands.log")]
+            logs = []
+            for source, path in files:
+                content, truncated = read_private(path, tail=True)
+                content = redact(content)
+                encoded = content.encode("utf-8")
+                if len(encoded) > LOG_BYTES:
+                    content = encoded[-LOG_BYTES:].decode("utf-8", errors="ignore")
+                    truncated = True
+                if content or truncated:
+                    logs.append({"source": source, "content": content, "truncated": truncated})
+            return {"requestId": request_id, "logs": logs}
+        except (OSError, ValueError, TypeError):
+            raise ApiError("LOGS_UNAVAILABLE", 503) from None
+
     def status(self):
         updater = self.factory(self.directory)
-        result = {"apiVersion": 1, "engine": public_state(updater.status()), "acceptingRequests": not self.closing}
+        result = {"apiVersion": 1, "engine": public_state(updater.status()), "acceptingRequests": not self.closing,
+                  "capabilities": {"requestLogs": True}}
         installed = engine.read_json(updater.home / "installed.json", {}).get("manifest", {})
         result["installed"] = {k: v for k, v in public_state({"version": installed.get("version"),
                               "revision": installed.get("commit")}).items() if k != "status"}
@@ -208,6 +316,7 @@ class Controller:
 
     def run(self, record, fingerprint):
         request_id = record["requestId"]
+        updater = None
         try:
             record.update(status="running", updatedAt=engine.now())
             engine.write_json(self.path(request_id), record)
@@ -224,6 +333,8 @@ class Controller:
                 os.chmod(log.name, 0o600)
                 traceback.print_exc(file=log)
         finally:
+            if updater and updater.state and updater.state.get("requestId") == request_id:
+                record["job"] = str(updater.job)
             record["updatedAt"] = engine.now()
             engine.write_json(self.path(request_id), record)
 
@@ -267,7 +378,9 @@ class Handler(BaseHTTPRequestHandler):
             if self.command == "GET" and self.path == "/v1/status":
                 self.send_json(200, controller.status())
             elif self.command == "GET" and self.path.startswith("/v1/requests/"):
-                self.send_json(200, controller.request_status(self.path.removeprefix("/v1/requests/")))
+                suffix = self.path.removeprefix("/v1/requests/")
+                self.send_json(200, controller.request_logs(suffix.removesuffix("/logs"))
+                               if suffix.endswith("/logs") else controller.request_status(suffix))
             elif self.command == "POST" and self.path in ("/v1/check", "/v1/apply"):
                 lengths = self.headers.get_all("Content-Length", [])
                 if (len(lengths) != 1 or not lengths[0].isdigit() or not 1 <= int(lengths[0]) <= 4096

scripts/update/tests/test_service.py

@@ -8,6 +8,7 @@ import subprocess
 import tempfile
 import threading
 import unittest
+from unittest.mock import patch
 
 ROOT = Path(__file__).resolve().parents[3]
 spec = importlib.util.spec_from_file_location('update_service', ROOT / 'scripts/selfhost-update-service.py')
@@ -232,6 +233,86 @@ class ServiceTests(unittest.TestCase):
         self.assertEqual(self.request('GET', '/v1/logs')[0], 404)
         self.assertEqual((self.controller.home / 'check-request-01.log').stat().st_mode & 0o777, 0o600)
 
+    def test_logs_include_failure_commands_and_events_with_credentials_redacted(self):
+        (self.root / '.env').write_text('POSTGRES_PASSWORD="database-secret"\nAPI_TOKEN=api-secret\n'
+                                      'SMTP_PASSWORD=complex,secret/suffix\n'
+                                      'DATABASE_URL=postgres://user:db%2Fsecret@db/fedeo\n')
+        (self.root / 'docker-compose.override.yml').write_text('  MINIO_ROOT_PASSWORD: "minio-secret"\n')
+        self.factory.failure = s.engine.UpdateError('Registry unauthorized: database-secret')
+        request_id = self.check()
+        job = self.root / '.updates' / ('job-' + request_id)
+        (job / 'target.env').write_text("SMTP_PASSWORD='previous-secret'\n")
+        (job / 'commands.log').write_text(
+            '\x1b[31mError: pull access denied\x1b[0m\n'
+            'database-secret api-secret minio-secret previous-secret release-secret\n'
+            'complex,secret/suffix complex%2Csecret%2Fsuffix db/secret\n'
+            'https://user:url-secret@registry.test/image\n'
+            'Authorization: Bearer header-secret\n'
+            '{"password": "json-secret", "access_token": "query-secret"}\n'
+            'TOKEN=unquoted-secret\n--password cli-secret\n')
+        with patch.dict(s.os.environ, {'FEDEO_RELEASE_TOKEN': 'release-secret'}):
+            status, result = self.request('GET', f'/v1/requests/{request_id}/logs')
+        self.assertEqual(status, 200)
+        self.assertEqual(result['requestId'], request_id)
+        logs = {entry['source']: entry['content'] for entry in result['logs']}
+        self.assertIn('Registry unauthorized', logs['service'])
+        self.assertIn('pull access denied', logs['commands'])
+        self.assertIn('validating', logs['events'])
+        for secret in ('database-secret', 'api-secret', 'minio-secret', 'previous-secret', 'release-secret',
+                       'url-secret', 'header-secret', 'json-secret', 'query-secret', 'unquoted-secret', 'cli-secret',
+                       'complex,secret/suffix', 'complex%2Csecret%2Fsuffix', 'db/secret', '\x1b'):
+            self.assertNotIn(secret, json.dumps(result))
+        self.assertTrue(self.request('GET', '/v1/status')[1]['capabilities']['requestLogs'])
+
+    def test_logs_survive_restart_and_find_legacy_jobs_without_crossing_requests(self):
+        first = self.check()
+        first_job = self.root / '.updates' / ('job-' + first)
+        (first_job / 'commands.log').write_text('first request only\n')
+        # Simulate the private request record from an older host service.
+        record = self.controller.read(first)
+        record.pop('job')
+        s.engine.write_json(self.controller.path(first), record)
+        second = self.check('check-request-02')
+        (self.root / '.updates' / ('job-' + second) / 'commands.log').write_text('second request only\n')
+        self.controller.close()
+        self.controller = s.Controller(self.root, updater_factory=self.factory)
+        self.server.controller = self.controller
+        for request_id, own, other in [(first, 'first', 'second'), (second, 'second', 'first')]:
+            payload = json.dumps(self.request('GET', f'/v1/requests/{request_id}/logs')[1])
+            self.assertIn(own + ' request only', payload)
+            self.assertNotIn(other + ' request only', payload)
+        self.assertEqual(self.request('GET', '/v1/requests/missing-request/logs')[0], 404)
+        self.assertEqual(self.request('GET', '/v1/requests/../../secret/logs')[0], 400)
+
+    def test_running_logs_are_bounded_and_do_not_expose_partial_secret_lines(self):
+        self.factory.release.clear()
+        request_id = 'check-running-logs'
+        self.request('POST', '/v1/check', {'requestId': request_id})
+        self.assertTrue(self.factory.started.wait(3))
+        job = self.root / '.updates' / ('job-' + request_id)
+        (job / 'commands.log').write_text('PASSWORD=' + 'secret-fragment' * 4000 + '\n' + 'old entry\n' * 900 + 'Last diagnostic: ä\n')
+        result = self.request('GET', f'/v1/requests/{request_id}/logs')[1]
+        commands = next(entry for entry in result['logs'] if entry['source'] == 'commands')
+        self.assertTrue(commands['truncated'])
+        self.assertLessEqual(len(commands['content'].encode()), s.LOG_BYTES)
+        self.assertIn('Last diagnostic: ä', commands['content'])
+        self.assertNotIn('secret-fragment', commands['content'])
+        self.assertLess(len(json.dumps(result).encode()), 65536)
+
+    def test_logs_fail_closed_for_symlinks_and_unreadable_redaction_configuration(self):
+        request_id = self.check()
+        job = self.root / '.updates' / ('job-' + request_id)
+        private = self.root / 'private.txt'
+        private.write_text('never publish this')
+        (job / 'commands.log').symlink_to(private)
+        status, result = self.request('GET', f'/v1/requests/{request_id}/logs')
+        self.assertEqual(status, 503)
+        self.assertEqual(result['error']['code'], 'LOGS_UNAVAILABLE')
+        self.assertNotIn('never publish this', json.dumps(result))
+        (job / 'commands.log').unlink()
+        (self.root / '.env').symlink_to(private)
+        self.assertEqual(self.request('GET', f'/v1/requests/{request_id}/logs')[0], 503)
+
     def test_api_rejects_unsupported_commands_options_bodies_and_browser_requests(self):
         for body in [[], {'requestId': 'bad/path'}, {'requestId': 'check-request-01', 'version': ';reboot'},
                      {'requestId': 'check-request-01', 'directory': '/etc'},