437 lines
21 KiB
Python
437 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""Closed-file H2 upgrades with a durable recovery journal. Hostsharing layout."""
|
|
import fcntl
|
|
import hashlib
|
|
import http.cookiejar
|
|
import html
|
|
import urllib.parse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid
|
|
import zipfile
|
|
|
|
|
|
def atomic(path, value):
|
|
tmp = path.with_name(path.name + '.next')
|
|
with tmp.open('w') as stream:
|
|
stream.write(value)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
tmp.chmod(0o600)
|
|
tmp.replace(path)
|
|
sync_directory(path.parent)
|
|
|
|
|
|
def sync_directory(path):
|
|
fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY)
|
|
try:
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def durable_tree(folder):
|
|
for path in folder.rglob('*'):
|
|
if path.is_file():
|
|
with path.open('rb') as stream:
|
|
os.fsync(stream.fileno())
|
|
for path in sorted((p for p in folder.rglob('*') if p.is_dir()), reverse=True):
|
|
sync_directory(path)
|
|
sync_directory(folder)
|
|
|
|
|
|
def replace_database(source, target, displaced):
|
|
"""Shared restore core: prepare a durable copy before displacing the closed DB."""
|
|
staged = target.parent / ('restore-data-' + uuid.uuid4().hex)
|
|
try:
|
|
shutil.copytree(source, staged)
|
|
durable_tree(staged)
|
|
if target.exists():
|
|
target.rename(displaced / ('displaced-data-' + uuid.uuid4().hex))
|
|
sync_directory(displaced)
|
|
sync_directory(target.parent)
|
|
staged.rename(target)
|
|
sync_directory(target.parent)
|
|
finally:
|
|
if staged.exists():
|
|
shutil.rmtree(staged)
|
|
|
|
|
|
def link(path, target):
|
|
tmp = path.with_name(path.name + '.next')
|
|
if tmp.is_symlink():
|
|
tmp.unlink()
|
|
tmp.symlink_to(target)
|
|
tmp.replace(path)
|
|
sync_directory(path.parent)
|
|
|
|
|
|
def checksum(path):
|
|
with path.open('rb') as stream:
|
|
return hashlib.file_digest(stream, 'sha256').hexdigest()
|
|
|
|
|
|
def manifest(folder):
|
|
value = json.loads((folder / 'manifest.json').read_text())
|
|
digest = checksum(folder / 'werkjournal.jar')
|
|
if (value.get('schemaVersion') != 1 or value.get('runtime') != 'java-25'
|
|
or not re.fullmatch('[0-9a-f]{40}', value.get('commit', ''))
|
|
or value.get('sha256') != digest or value.get('files') != {'werkjournal.jar': digest}):
|
|
raise ValueError('Invalid release manifest or checksum')
|
|
return value
|
|
|
|
|
|
class Deployment:
|
|
service = 'werkjournal-backend.service'
|
|
url = 'https://werkjournal.javagil.de'
|
|
|
|
def __init__(self, home):
|
|
self.home = home
|
|
self.app = home / 'opt/werkjournal'
|
|
self.site = home / 'doms/werkjournal.javagil.de/htdocs-ssl'
|
|
self.flag = self.site / 'maintenance.flag'
|
|
self.pending = self.app / 'pending-deployment.json'
|
|
|
|
def command(self, *args):
|
|
return subprocess.check_output(args, text=True, stderr=subprocess.PIPE).strip()
|
|
|
|
def stop(self):
|
|
self.command('systemctl', '--user', 'stop', self.service)
|
|
if self.command('systemctl', '--user', 'show', self.service, '--property=ActiveState', '--value') not in {'inactive', 'failed'}:
|
|
raise RuntimeError('Service did not stop; no database copy is safe')
|
|
if self.command('systemctl', '--user', 'show', self.service, '--property=MainPID', '--value') != '0':
|
|
raise RuntimeError('Service still has a process')
|
|
|
|
def start(self):
|
|
self.command('systemctl', '--user', 'start', self.service)
|
|
|
|
def ready_once(self, expected):
|
|
# Only this loopback probe may transmit Secure cookies over HTTP. Production
|
|
# browser cookies remain Secure; TLS is terminated by the external proxy.
|
|
class LoopbackCookies(http.cookiejar.DefaultCookiePolicy):
|
|
def return_ok_secure(self, cookie, request):
|
|
return request.host == '127.0.0.1:18090'
|
|
jar = http.cookiejar.CookieJar(policy=LoopbackCookies())
|
|
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
|
|
base = 'http://127.0.0.1:18090'
|
|
logged_in = False
|
|
|
|
def csrf():
|
|
with opener.open(base + '/login', timeout=5) as response:
|
|
page = response.read().decode()
|
|
token = re.search(r'name="_csrf" value="([^"]+)"', page)
|
|
if not token:
|
|
raise ValueError('Login page has no CSRF token')
|
|
return urllib.parse.urlencode({'_csrf': html.unescape(token.group(1))}).encode()
|
|
|
|
try:
|
|
with opener.open(base + '/api/v1/info', timeout=3) as response:
|
|
needs_login = urllib.parse.urlsplit(response.url).path == '/login'
|
|
info = None if needs_login else json.load(response)
|
|
if not needs_login:
|
|
if info != {'application': 'werkjournal', 'version': expected['version']}:
|
|
return False
|
|
with opener.open(base + '/?v-r=init&location=&query=', timeout=5) as response:
|
|
needs_login = urllib.parse.urlsplit(response.url).path == '/login'
|
|
if not needs_login:
|
|
return ('application/json' in response.headers.get('Content-Type', '')
|
|
and isinstance(json.load(response).get('appConfig'), dict))
|
|
if needs_login:
|
|
with opener.open(base + '/login/guest', data=csrf(), timeout=5) as response:
|
|
response.read()
|
|
logged_in = True
|
|
try:
|
|
with opener.open(base + '/api/v1/info', timeout=3) as response:
|
|
info = json.load(response)
|
|
except urllib.error.HTTPError as error:
|
|
if error.code != 403:
|
|
raise
|
|
# Historical authenticated releases denied this route even to
|
|
# logged-in users. Verify their packaged version for rollback.
|
|
with zipfile.ZipFile(self.app / 'current/werkjournal.jar') as jarfile:
|
|
properties = jarfile.read('META-INF/build-info.properties').decode()
|
|
version = next(line.removeprefix('build.version=') for line in properties.splitlines()
|
|
if line.startswith('build.version='))
|
|
info = {'application': 'werkjournal', 'version': version}
|
|
if info != {'application': 'werkjournal', 'version': expected['version']}:
|
|
return False
|
|
with opener.open(base + '/?v-r=init&location=&query=', timeout=5) as response:
|
|
return ('application/json' in response.headers.get('Content-Type', '')
|
|
and isinstance(json.load(response).get('appConfig'), dict))
|
|
finally:
|
|
if logged_in:
|
|
with opener.open(base + '/logout', data=csrf(), timeout=5) as response:
|
|
response.read()
|
|
|
|
def ready(self, expected):
|
|
deadline = time.monotonic() + 90
|
|
while time.monotonic() < deadline:
|
|
try:
|
|
if self.ready_once(expected):
|
|
return
|
|
except (OSError, ValueError):
|
|
pass
|
|
time.sleep(2)
|
|
raise RuntimeError('Application or Flow initialization failed')
|
|
|
|
def preflight(self, check_data=True, allow_restore=False):
|
|
if not allow_restore and (self.app / 'pending-restore.json').exists():
|
|
raise RuntimeError('Pending operator restore: run remote backend restore --recover first')
|
|
if not self.app.is_dir() or not self.site.is_dir():
|
|
raise RuntimeError('Expected Hostsharing directories are missing')
|
|
# The initial installation uses this closed-file database. Refuse runtime
|
|
# overrides until their backup paths have been explicitly configured.
|
|
unit = self.home / '.config/systemd/user/werkjournal-backend.service'
|
|
content = unit.read_text()
|
|
lines = [line.strip() for line in content.splitlines() if line.strip() and not line.startswith(('#', '['))]
|
|
expected_start = f'ExecStart={self.home}/opt/jdk25/bin/java -Xms32m -Xmx384m -XX:MaxMetaspaceSize=192m -XX:ActiveProcessorCount=2 -Djava.awt.headless=true -jar {self.app}/current/werkjournal.jar'
|
|
if ([line for line in lines if line.startswith('WorkingDirectory=')] != [f'WorkingDirectory={self.app}']
|
|
or [line for line in lines if line.startswith('ExecStart=')] != [expected_start]):
|
|
raise RuntimeError('Unexpected service paths or JVM options')
|
|
allowed = {'Description', 'After', 'Type', 'WorkingDirectory', 'ExecStart', 'Environment', 'EnvironmentFile', 'Restart', 'RestartSec', 'TimeoutStopSec', 'UMask', 'WantedBy'}
|
|
if any(line.partition('=')[0] not in allowed for line in lines):
|
|
raise RuntimeError('Unexpected service directives require review')
|
|
expected_environment = {'Environment=BACKEND_PORT=18090', 'Environment=SERVER_ADDRESS=127.0.0.1',
|
|
'Environment=SERVER_FORWARD_HEADERS_STRATEGY=native', 'Environment=SERVER_TOMCAT_REMOTEIP_HOST_HEADER=x-forwarded-host'}
|
|
if set(line for line in lines if line.startswith('Environment=')) != expected_environment:
|
|
raise RuntimeError('Unexpected inline environment')
|
|
if [line for line in lines if line.startswith('EnvironmentFile=')] != ['EnvironmentFile=-%h/.config/werkjournal/environment']:
|
|
raise RuntimeError('Unexpected environment file configuration')
|
|
environment_file = self.home / '.config/werkjournal/environment'
|
|
if environment_file.exists():
|
|
if environment_file.stat().st_mode & 0o077:
|
|
raise RuntimeError('Runtime environment file must be private to its owner')
|
|
for line in environment_file.read_text().splitlines():
|
|
parts = shlex.split(line, comments=True)
|
|
if not parts:
|
|
continue
|
|
if len(parts) != 1 or '=' not in parts[0]:
|
|
raise RuntimeError('Unsupported runtime environment file syntax')
|
|
key, value = parts[0].split('=', 1)
|
|
if key in {'WERKJOURNAL_DATABASE_URL', 'SPRING_DATASOURCE_URL'}:
|
|
if value != 'jdbc:h2:file:./data/werkjournal':
|
|
raise RuntimeError('Database override is outside the managed backup path')
|
|
elif key == 'BACKEND_PORT':
|
|
if value != '18090':
|
|
raise RuntimeError('Unexpected backend port override')
|
|
elif not (key.startswith('WERKJOURNAL_') or key.startswith('SPRING_SECURITY_OAUTH2_CLIENT_')):
|
|
raise RuntimeError('Unsupported runtime environment key: ' + key)
|
|
for prop, expected in [('DropInPaths', ''), ('NeedDaemonReload', 'no')]:
|
|
if self.command('systemctl', '--user', 'show', self.service, f'--property={prop}', '--value') != expected:
|
|
raise RuntimeError('Unexpected service overrides or pending unit reload')
|
|
environment = self.command('systemctl', '--user', 'show-environment')
|
|
if any(line.startswith(('SPRING_', 'WERKJOURNAL_DATABASE_', 'JAVA_TOOL_OPTIONS=', 'JDK_JAVA_OPTIONS=', '_JAVA_OPTIONS=')) for line in environment.splitlines()):
|
|
raise RuntimeError('Manager environment may override the database configuration')
|
|
if not check_data:
|
|
return
|
|
if not (self.app / 'data/werkjournal.mv.db').is_file():
|
|
raise RuntimeError('Expected H2 file is missing')
|
|
if (self.app / 'data').is_symlink():
|
|
raise RuntimeError('Database directory must not be a symlink')
|
|
for path in (self.app / 'data').rglob('*'):
|
|
if path.is_symlink() or not path.is_file():
|
|
raise RuntimeError('Unexpected database directory layout')
|
|
|
|
def stage(self, stream):
|
|
stage = Path(tempfile.mkdtemp(prefix='candidate-', dir=self.app))
|
|
try:
|
|
seen = set()
|
|
with tarfile.open(fileobj=stream, mode='r|') as archive:
|
|
for entry in archive:
|
|
if entry.name not in {'manifest.json', 'werkjournal.jar'} or entry.name in seen or not entry.isfile() or entry.size > 300 * 1024 * 1024:
|
|
raise ValueError('Invalid artifact archive')
|
|
seen.add(entry.name)
|
|
with archive.extractfile(entry) as source, (stage / entry.name).open('wb') as output:
|
|
shutil.copyfileobj(source, output)
|
|
output.flush()
|
|
os.fsync(output.fileno())
|
|
if len(seen) != 2:
|
|
raise ValueError('Incomplete artifact archive')
|
|
value = manifest(stage)
|
|
release = self.app / 'releases' / value['commit']
|
|
if release.exists():
|
|
if manifest(release) != value:
|
|
raise ValueError('Release ID already has different contents')
|
|
else:
|
|
stage.rename(release)
|
|
sync_directory(release.parent)
|
|
return value
|
|
finally:
|
|
if stage.exists():
|
|
shutil.rmtree(stage)
|
|
|
|
def upload(self, stream):
|
|
self.preflight()
|
|
if self.pending.exists():
|
|
raise RuntimeError('Resolve the pending deployment before uploading another candidate')
|
|
candidate = self.stage(stream)
|
|
return {'result': 'STAGED', 'commit': candidate['commit'],
|
|
'message': 'Candidate stored; deploy is required for activation'}
|
|
|
|
def journal(self, state, phase):
|
|
state['phase'] = phase
|
|
atomic(self.pending, json.dumps(state))
|
|
|
|
def maintenance(self):
|
|
atomic(self.flag, 'Wartungsmodus — bitte in wenigen Minuten erneut versuchen.\n')
|
|
try:
|
|
urllib.request.urlopen(self.url + '/', timeout=15)
|
|
raise RuntimeError('Public maintenance did not return HTTP 503')
|
|
except urllib.error.HTTPError as error:
|
|
if error.code != 503 or 'Wartungsmodus' not in error.read().decode():
|
|
raise
|
|
|
|
def restore(self, state):
|
|
try:
|
|
self._restore(state)
|
|
except BaseException:
|
|
try:
|
|
self.stop()
|
|
except Exception:
|
|
pass
|
|
raise
|
|
|
|
def _restore(self, state):
|
|
self.stop()
|
|
bundle = self.app / 'rollbacks' / state['backup']
|
|
if state['phase'] in {'backed_up', 'switched', 'verified'}:
|
|
inventory = json.loads((bundle / 'inventory.json').read_text())
|
|
if 'werkjournal.mv.db' not in inventory or any(Path(name).name != name for name in inventory):
|
|
raise RuntimeError('Invalid backup inventory')
|
|
if any(path.is_symlink() or not path.is_file() for path in (bundle / 'data').iterdir()):
|
|
raise RuntimeError('Invalid backup layout')
|
|
for name, digest in inventory.items():
|
|
if checksum(bundle / 'data' / name) != digest:
|
|
raise RuntimeError('Backup checksum mismatch; maintenance stays active')
|
|
replace_database(bundle / 'data', self.app / 'data', bundle)
|
|
old = self.app / 'releases' / state['old']['commit']
|
|
if manifest(old) != state['old']:
|
|
raise RuntimeError('Previous application failed verification')
|
|
link(self.app / 'current', old.relative_to(self.app))
|
|
atomic(self.app / 'deployed.json', json.dumps(state['old'], indent=2) + '\n')
|
|
self.start()
|
|
self.ready(state['old'])
|
|
self.journal(state, 'restored')
|
|
self.flag.unlink()
|
|
sync_directory(self.site)
|
|
self.pending.unlink()
|
|
sync_directory(self.app)
|
|
|
|
def recover(self):
|
|
self.preflight(check_data=False)
|
|
state = json.loads(self.pending.read_text())
|
|
if (not re.fullmatch('[0-9a-f]{32}', state.get('backup', ''))
|
|
or any(not re.fullmatch('[0-9a-f]{40}', state.get(key, {}).get('commit', '')) for key in ['old', 'new'])):
|
|
raise RuntimeError('Invalid recovery journal')
|
|
if state['phase'] == 'published':
|
|
self.ready(state['new'])
|
|
if self.flag.exists():
|
|
self.flag.unlink()
|
|
sync_directory(self.site)
|
|
self.finalize(state)
|
|
return
|
|
if not self.flag.exists():
|
|
if state['phase'] in {'prepared', 'restored'}:
|
|
self.pending.unlink()
|
|
return
|
|
if state['phase'] != 'verified':
|
|
raise RuntimeError('Unexpected missing maintenance flag; manual inspection required')
|
|
# Publication may have happened before the process died. Never undo
|
|
# writes accepted since then: finish bookkeeping, not rollback.
|
|
self.finalize(state)
|
|
return
|
|
self.restore(state)
|
|
|
|
def finalize(self, state):
|
|
link(self.app / 'previous', Path('releases') / state['old']['commit'])
|
|
link(self.app / 'rollback', Path('rollbacks') / state['backup'])
|
|
self.pending.unlink(missing_ok=True)
|
|
sync_directory(self.app)
|
|
# Only deployment-owned UUID bundles and full-commit release directories
|
|
# are eligible; unrelated operator files are never swept.
|
|
for folder, retained, pattern in [
|
|
(self.app / 'rollbacks', {state['backup']}, '[0-9a-f]{32}'),
|
|
(self.app / 'releases', {state['old']['commit'], state['new']['commit']}, '[0-9a-f]{40}')]:
|
|
for path in folder.iterdir():
|
|
if path.name not in retained and re.fullmatch(pattern, path.name) and path.is_dir() and not path.is_symlink():
|
|
try:
|
|
shutil.rmtree(path)
|
|
except OSError:
|
|
print('Old deployment artifact cleanup deferred', file=sys.stderr)
|
|
|
|
def deploy(self, stream):
|
|
self.preflight()
|
|
if self.pending.exists() or self.flag.exists():
|
|
raise RuntimeError('Unresolved maintenance/deployment: use recovery first')
|
|
old = json.loads((self.app / 'deployed.json').read_text())
|
|
if not re.fullmatch('[0-9a-f]{40}', old.get('commit', '')):
|
|
raise RuntimeError('Invalid deployed commit')
|
|
old_release = self.app / 'releases' / old['commit']
|
|
if (self.app / 'current').resolve() != old_release or manifest(old_release) != old:
|
|
raise RuntimeError('Current release differs from its verified manifest')
|
|
new = self.stage(stream)
|
|
if new['commit'] == old['commit']:
|
|
return {'result': 'UNCHANGED', 'commit': new['commit']}
|
|
if old['commit'] not in new.get('ancestors', []):
|
|
raise RuntimeError('Candidate does not descend from the deployed commit')
|
|
state = {'old': old, 'new': new, 'backup': uuid.uuid4().hex}
|
|
self.journal(state, 'prepared')
|
|
try:
|
|
self.maintenance()
|
|
self.stop()
|
|
bundle = self.app / 'rollbacks' / state['backup']
|
|
bundle.mkdir(parents=True, mode=0o700)
|
|
shutil.copytree(self.app / 'data', bundle / 'data')
|
|
inventory = {}
|
|
for path in (bundle / 'data').iterdir():
|
|
with path.open('rb') as stream:
|
|
os.fsync(stream.fileno())
|
|
inventory[path.name] = checksum(path)
|
|
atomic(bundle / 'inventory.json', json.dumps(inventory))
|
|
atomic(bundle / 'manifest.json', json.dumps(old))
|
|
sync_directory(bundle / 'data')
|
|
sync_directory(bundle.parent)
|
|
self.journal(state, 'backed_up')
|
|
link(self.app / 'current', Path('releases') / new['commit'])
|
|
self.journal(state, 'switched')
|
|
self.start()
|
|
self.ready(new)
|
|
atomic(self.app / 'deployed.json', json.dumps(new, indent=2) + '\n')
|
|
self.journal(state, 'verified')
|
|
except Exception:
|
|
if self.flag.exists():
|
|
self.restore(state)
|
|
raise
|
|
# Durably record publication intent BEFORE exposing the application.
|
|
# Recovery must preserve new writes even if the flag reappears on reboot.
|
|
self.journal(state, 'published')
|
|
self.flag.unlink()
|
|
sync_directory(self.site)
|
|
self.finalize(state)
|
|
return {'result': 'PASS', 'commit': new['commit'], 'url': self.url}
|
|
|
|
|
|
if __name__ == '__main__':
|
|
os.umask(0o077)
|
|
deployment = Deployment(Path.home())
|
|
with (deployment.app / '.deploy.lock').open('a') as lock:
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
if '--upload' in sys.argv:
|
|
print(json.dumps(deployment.upload(sys.stdin.buffer)))
|
|
elif '--recover' in sys.argv:
|
|
deployment.recover()
|
|
print(json.dumps({'result': 'RECOVERED'}))
|
|
else:
|
|
print(json.dumps(deployment.deploy(sys.stdin.buffer)))
|