89 lines
3.8 KiB
Python
89 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Read server metadata for migration without exposing config values or secrets."""
|
|
import configparser
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
def cron_units(nc_dir):
|
|
"""Discover directly configured occ/cron.php writers, not arbitrary wrappers."""
|
|
proc = subprocess.run(['systemctl','list-unit-files','--type=service','--no-legend','--no-pager'], check=True, text=True, capture_output=True)
|
|
services = []
|
|
for line in proc.stdout.splitlines():
|
|
unit = line.split()[0]
|
|
# Templates cannot be stopped without a concrete instance name.
|
|
if '@.' in unit:
|
|
continue
|
|
command = subprocess.check_output(['systemctl','show',unit,'--property=ExecStart','--value'],text=True)
|
|
if nc_dir in command and ('cron.php' in command or '/occ' in command):
|
|
services.append(unit)
|
|
timers = []
|
|
for service in services:
|
|
triggered = subprocess.check_output(['systemctl','show',service,'--property=TriggeredBy','--value'],text=True)
|
|
timers.extend(x for x in triggered.split() if x.endswith(('.timer','.socket')))
|
|
for unit in dict.fromkeys(timers + services):
|
|
if not re.fullmatch(r'[A-Za-z0-9_.@:-]+\.(service|timer|socket)', unit):
|
|
raise ValueError('Unexpected systemd unit name')
|
|
print(unit)
|
|
|
|
|
|
def check_crontabs(nc_dir, nc_user):
|
|
"""Refuse to guess scheduling semantics for cron scripts; report locations only."""
|
|
found = []
|
|
paths = [Path('/etc/crontab')]
|
|
for directory in ('/etc/cron.d','/etc/cron.hourly','/etc/cron.daily','/etc/cron.weekly','/etc/cron.monthly','/var/spool/cron'):
|
|
root = Path(directory)
|
|
if root.is_dir():
|
|
paths.extend(p for p in root.rglob('*') if p.is_file())
|
|
for path in paths:
|
|
if not path.is_file():
|
|
continue
|
|
for line in path.read_text(errors='replace').splitlines():
|
|
if not line.lstrip().startswith('#') and (nc_dir in line or 'cron.php' in line):
|
|
found.append(str(path))
|
|
break
|
|
if found:
|
|
print('Nextcloud cron entries: '+', '.join(sorted(set(found))))
|
|
sys.exit(2)
|
|
|
|
|
|
def gitea(path):
|
|
parser = configparser.ConfigParser(interpolation=None, strict=False)
|
|
parser.read_string('[DEFAULT]\n' + Path(path).read_text())
|
|
result = {}
|
|
# Only explicitly configured absolute filesystem paths are accepted here.
|
|
for key, section, option in [('GITEA_DATA_DIR','DEFAULT','APP_DATA_PATH'),('GITEA_REPOS_DIR','repository','ROOT')]:
|
|
value = parser[section].get(option,'') if section in parser else ''
|
|
if value and os.path.isabs(value) and '%' not in value:
|
|
result[key] = value
|
|
if parser.get('database','DB_TYPE',fallback='') == 'mysql':
|
|
result['GITEA_DB_NAME'] = parser.get('database','NAME',fallback='')
|
|
for key,value in result.items():
|
|
print(key+'='+shlex.quote(value))
|
|
|
|
|
|
def environment_config():
|
|
raw = subprocess.check_output(['systemctl','show','app-backup.service','--property=Environment','--value'],text=True)
|
|
for item in shlex.split(raw):
|
|
if item.startswith('RCLONE_CONFIG='):
|
|
print(item.split('=',1)[1])
|
|
return
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
op,*args=sys.argv[1:]
|
|
if op == 'cron-units': cron_units(*args)
|
|
elif op == 'check-crontabs': check_crontabs(*args)
|
|
elif op == 'gitea': gitea(*args)
|
|
elif op == 'rclone-config': environment_config()
|
|
else: raise ValueError('Unknown discovery operation')
|
|
except (ValueError, OSError, configparser.Error, subprocess.CalledProcessError):
|
|
# Parser errors can include secret config lines, so do not echo exceptions.
|
|
print('ERROR: Discovery incomplete. Check Nextcloud cron scheduling or explicit Gitea/rclone settings; no secret values logged.',file=sys.stderr)
|
|
sys.exit(1)
|