mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-01 10:22:08 +00:00
252 lines
13 KiB
Python
252 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Proxmox VE and Proxmox Backup Server repository policy via their own parsed APT
|
|
repository API.
|
|
|
|
Only `notfound` or `expired` permits an offered switch. Never print raw subscription or
|
|
repository API responses: they can contain subscription keys or credentials.
|
|
"""
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import string
|
|
import sys
|
|
|
|
ENDPOINT = '/nodes/localhost/apt/repositories'
|
|
PVE_CHANNELS = {'pve-enterprise', 'pve-no-subscription', 'pve-test', 'pvetest'}
|
|
# Proxmox VE and Proxmox Backup Server share the repository API; each has its
|
|
# own command line, channels and place in the web interface.
|
|
PRODUCTS = {
|
|
'pve': {'api': ['pvesh'], 'name': 'Proxmox VE', 'short': 'PVE',
|
|
'channels': PVE_CHANNELS, 'enterprise': 'pve-enterprise', 'public': 'pve-no-subscription',
|
|
'test': {'pve-no-subscription', 'pve-test', 'pvetest'},
|
|
'ui': 'Node > Updates > Repositories', 'ceph': True},
|
|
'pbs': {'api': ['proxmox-backup-debug', 'api'], 'name': 'Proxmox Backup Server', 'short': 'PBS',
|
|
'channels': {'pbs-enterprise', 'pbs-no-subscription', 'pbs-test', 'pbstest'},
|
|
'enterprise': 'pbs-enterprise', 'public': 'pbs-no-subscription',
|
|
'test': {'pbs-no-subscription', 'pbs-test', 'pbstest'},
|
|
'ui': 'Administration > Repositories', 'ceph': False},
|
|
}
|
|
BASE = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
def N_(text):
|
|
"""Mark a CLI message without changing the policy's English exception."""
|
|
return text
|
|
|
|
|
|
class PolicyError(ValueError):
|
|
def __init__(self, template, **params):
|
|
self.template = template
|
|
self.params = params
|
|
super().__init__(template.format(**params))
|
|
|
|
|
|
def cli_error(error, product):
|
|
"""Localize only PBS CLI diagnostics; policy callers retain English errors."""
|
|
if product != 'pbs' or not isinstance(error, PolicyError):
|
|
return str(error)
|
|
try:
|
|
language = json.loads((BASE / 'config.json').read_text()).get('language', 'en')
|
|
if language not in ('sk', 'de', 'es', 'fr', 'it', 'pt', 'sv'):
|
|
return str(error)
|
|
catalog = json.loads((BASE / 'lang' / f'{language}.json').read_text())
|
|
message = catalog.get(error.template)
|
|
if isinstance(message, str) and message.strip():
|
|
fields = lambda text: {name for _, name, _, _ in string.Formatter().parse(text) if name is not None}
|
|
if fields(message) != fields(error.template) or any(
|
|
spec or conversion for _, _, spec, conversion in string.Formatter().parse(message)
|
|
):
|
|
return str(error)
|
|
return message.format(**error.params)
|
|
except (OSError, ValueError, KeyError, IndexError, TypeError, AttributeError):
|
|
pass
|
|
return str(error)
|
|
|
|
|
|
def run(args):
|
|
try:
|
|
return subprocess.run(args, check=True, capture_output=True, text=True).stdout
|
|
except (OSError, subprocess.CalledProcessError) as exc:
|
|
raise ValueError(f'Unable to query or change Proxmox repository state ({args[0]}). Check the service and retry.') from exc
|
|
|
|
|
|
def subscription_status(product='pve'):
|
|
if product == 'pbs':
|
|
# Only the status field is read; the answer also carries the server ID.
|
|
try:
|
|
status = json.loads(run(['proxmox-backup-manager', 'subscription', 'get', '--output-format', 'json'])).get('status')
|
|
except (ValueError, AttributeError):
|
|
status = None
|
|
if status not in ('active', 'notfound', 'expired'):
|
|
raise PolicyError(N_('Subscription status is not unambiguously active, notfound or expired. Check proxmox-backup-manager subscription get locally; no repository changed.'))
|
|
return status
|
|
# pvesubscription's CLI get printer emits sorted "key: value" lines,
|
|
# including a secret key and server ID. Only parse the exact status line.
|
|
output = run(['pvesubscription', 'get'])
|
|
statuses = re.findall(r'^status: ([a-z]+)$', output, re.MULTILINE)
|
|
if len(statuses) != 1 or statuses[0] not in ('active', 'notfound', 'expired'):
|
|
raise ValueError('Subscription status is not unambiguously active, notfound or expired. Check pvesubscription get locally; no repository changed.')
|
|
return statuses[0]
|
|
|
|
|
|
def repository_state(suite, product='pve'):
|
|
p = PRODUCTS[product]
|
|
try:
|
|
data = json.loads(run([*p['api'], 'get', ENDPOINT, '--output-format', 'json']))
|
|
if not isinstance(data, dict) or not isinstance(data['files'], list) or not isinstance(data['errors'], list) or not isinstance(data['digest'], str) or not isinstance(data['standard-repos'], list):
|
|
raise ValueError()
|
|
if data['errors']:
|
|
raise ValueError()
|
|
entries = []
|
|
for file in data['files']:
|
|
for index, row in enumerate(file['repositories']):
|
|
if not isinstance(row, dict):
|
|
raise ValueError()
|
|
# Flat repositories (e.g. suite './') legitimately omit this.
|
|
row.setdefault('Components', [])
|
|
# Captured PVE 9.2 JSON uses integer 0/1; older fixtures use bool.
|
|
# Reject float/string coercions and retain the authoritative state.
|
|
if type(row['Enabled']) not in (bool, int) or row['Enabled'] not in (0, 1):
|
|
raise ValueError()
|
|
row['Enabled'] = bool(row['Enabled'])
|
|
if not all(
|
|
isinstance(row[k], list) and all(isinstance(value, str) for value in row[k])
|
|
for k in ('Types', 'URIs', 'Suites', 'Components')
|
|
):
|
|
raise ValueError()
|
|
# deb822 write/readback can insert Options/Enabled. It is
|
|
# redundant only when well-formed and consistent with Enabled;
|
|
# never discard the authoritative field or unrelated options.
|
|
options = row.get('Options', [])
|
|
if not isinstance(options, list):
|
|
raise ValueError()
|
|
remaining = []
|
|
seen_enabled = False
|
|
boolean_options = {'true': True, 'yes': True, '1': True,
|
|
'false': False, 'no': False, '0': False}
|
|
for option in options:
|
|
if not isinstance(option, dict) or not isinstance(option['Key'], str) \
|
|
or not isinstance(option['Values'], list) \
|
|
or not all(isinstance(value, str) for value in option['Values']):
|
|
raise ValueError()
|
|
if option['Key'].lower() != 'enabled':
|
|
remaining.append(option)
|
|
continue
|
|
if seen_enabled or set(option) != {'Key', 'Values'} or len(option['Values']) != 1:
|
|
raise ValueError()
|
|
value = boolean_options.get(option['Values'][0].lower())
|
|
if value is None or value != row['Enabled']:
|
|
raise ValueError()
|
|
seen_enabled = True
|
|
if remaining:
|
|
row['Options'] = remaining
|
|
else:
|
|
row.pop('Options', None)
|
|
entries.append((file['path'], index, row))
|
|
return data, entries
|
|
except (KeyError, TypeError, ValueError, IndexError) as exc:
|
|
raise PolicyError(N_('Proxmox APT repository inventory is invalid or reports parse errors. Fix sources in {ui}; no repository changed.'), ui=p['ui']) from exc
|
|
|
|
|
|
def evaluate(suite, apply=False, product='pve'):
|
|
if suite not in ('bookworm', 'trixie'):
|
|
raise PolicyError(N_('Unsupported Proxmox suite; no repository changed.'))
|
|
p = PRODUCTS[product]
|
|
data, entries = repository_state(suite, product)
|
|
pve = []
|
|
ceph = []
|
|
debian = False
|
|
wrong_pve = []
|
|
wrong_ceph = []
|
|
for path, index, row in entries:
|
|
if not row['Enabled'] or 'deb' not in row['Types']:
|
|
continue
|
|
components = set(row['Components'])
|
|
if components & p['channels']:
|
|
if suite not in row['Suites']:
|
|
wrong_pve.append(f"{' '.join(sorted(components & p['channels']))} in {path} (suite {' '.join(row['Suites'])})")
|
|
pve.append((path, index, row))
|
|
if 'main' in components and suite in row['Suites'] and not components & p['channels']:
|
|
debian = True
|
|
if p['ceph'] and 'enterprise' in components and any('/ceph-' in uri for uri in row['URIs']):
|
|
if suite not in row['Suites']:
|
|
wrong_ceph.append(f"{path} (suite {' '.join(row['Suites'])})")
|
|
ceph.append((path, index, row))
|
|
# Name the stale sources, never their URIs: those can carry credentials.
|
|
if wrong_pve:
|
|
raise PolicyError(N_('This host runs {name} on {suite}, but these {name} repositories are for another suite: {details}. Disable or remove them in {ui} and run the update again; no repository changed.'),
|
|
name=p['name'], suite=suite, details='; '.join(wrong_pve), ui=p['ui'])
|
|
if wrong_ceph:
|
|
raise ValueError(f"This host runs Proxmox VE on {suite}, but these Enterprise Ceph repositories are for another suite: "
|
|
f"{'; '.join(wrong_ceph)}. Correct them in Node > Updates > Repositories before switching; no repository changed.")
|
|
|
|
if any(set(row['Components']) & p['test'] for _, _, row in pve):
|
|
return 'preserve'
|
|
# Entitlement lookup is only needed when considering a switch. Existing
|
|
# public/test channels are the operator's choice, even if lookup fails.
|
|
status = subscription_status(product) # fail closed before any write
|
|
if status == 'active':
|
|
if not pve:
|
|
raise PolicyError(N_('Host has an active subscription but no active {short} repository. Configure its Enterprise source in {ui}; no repository changed.'), short=p['short'], ui=p['ui'])
|
|
return 'preserve'
|
|
if not debian:
|
|
raise PolicyError(N_('No active Debian base repository for this suite; configure it in the Proxmox repository UI before continuing.'))
|
|
# No active product source or only inaccessible Enterprise. A mixed stanza
|
|
# cannot be disabled without also disabling an unrelated component.
|
|
disable = [item for item in pve if p['enterprise'] in item[2]['Components']] + ceph
|
|
for _, _, row in disable:
|
|
expected = {p['enterprise']} if p['enterprise'] in row['Components'] else {'enterprise'}
|
|
if set(row['Components']) != expected:
|
|
raise PolicyError(N_('Enterprise shares an APT stanza with other components. Split it in the Proxmox repository UI; no repository changed.'))
|
|
handles = [r.get('handle') for r in data['standard-repos'] if r.get('handle') == 'no-subscription']
|
|
if len(handles) != 1:
|
|
raise PolicyError(N_('Proxmox no-subscription standard repository handle unavailable; no repository changed.'))
|
|
if not apply:
|
|
return 'offer'
|
|
# Adopt a refreshed digest only after checking the complete expected
|
|
# inventory: path/index targets are unsafe if another writer changed it.
|
|
expected = copy.deepcopy(data['files'])
|
|
try:
|
|
for path, index, original in disable:
|
|
next(file for file in expected if file['path'] == path)['repositories'][index]['Enabled'] = False
|
|
run([*p['api'], 'create', ENDPOINT, '--path', path, '--index', str(index),
|
|
'--enabled', '0', '--digest', data['digest']])
|
|
refreshed, current = repository_state(suite, product)
|
|
# Per-file digests change when Proxmox serializes the disabled entry.
|
|
inventory = lambda files: sorted(
|
|
({key: value for key, value in file.items() if key != 'digest'} for file in files),
|
|
key=lambda file: file['path'])
|
|
if inventory(refreshed['files']) != inventory(expected):
|
|
raise ValueError('Could not verify the expected repository inventory after Enterprise disable; possible concurrent edit. Inspect the Proxmox repository UI before retrying.')
|
|
data = refreshed
|
|
run([*p['api'], 'set', ENDPOINT, '--handle', 'no-subscription', '--digest', data['digest']])
|
|
_, current = repository_state(suite, product)
|
|
if not any(row['Enabled'] and 'deb' in row['Types'] and suite in row['Suites']
|
|
and p['public'] in row['Components'] for _, _, row in current) \
|
|
or any(p == path and i == index and row['Enabled']
|
|
for path, index, _ in disable for p, i, row in current):
|
|
raise ValueError('Could not verify the switched repositories; inspect the Proxmox repository UI before retrying.')
|
|
except ValueError as exc:
|
|
raise PolicyError(N_('Could not complete or verify the repository switch; changes may be partial and repository state is unknown. Inspect {ui} before retrying; no automatic rollback was attempted.'), ui=p['ui']) from exc
|
|
return 'changed'
|
|
|
|
|
|
def main():
|
|
product = 'pve'
|
|
try:
|
|
if len(sys.argv) not in (3, 4) or sys.argv[1] not in ('plan', 'apply') \
|
|
or (len(sys.argv) == 4 and sys.argv[3] not in PRODUCTS):
|
|
raise ValueError('Usage: repository_policy.py plan|apply bookworm|trixie [pve|pbs]')
|
|
product = sys.argv[3] if len(sys.argv) == 4 else 'pve'
|
|
print(evaluate(sys.argv[2], apply=sys.argv[1] == 'apply', product=product))
|
|
except ValueError as exc:
|
|
print(cli_error(exc, product), file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|