Files
ansible/roles/arch_update/files/aur_diff_check.py
T

149 lines
5.3 KiB
Python
Executable File

#!/usr/bin/env python3
import urllib.request
import urllib.parse
import json
import subprocess
import os
import re
import tempfile
import sys
import shutil
def run_cmd(cmd, cwd=None):
return subprocess.run(cmd, cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
def get_local_foreign_packages():
res = run_cmd(['pacman', '-Qm'])
if res.returncode != 0:
return {}
pkgs = {}
for line in res.stdout.strip().split('\n'):
if line:
parts = line.split()
if len(parts) == 2:
pkgs[parts[0]] = parts[1]
return pkgs
def get_aur_info(pkgnames):
if not pkgnames:
return {}
results = {}
chunk_size = 100
pkgs = list(pkgnames)
for i in range(0, len(pkgs), chunk_size):
chunk = pkgs[i:i+chunk_size]
url = "https://aur.archlinux.org/rpc/v5/info?" + "&".join([f"arg[]={urllib.parse.quote(p)}" for p in chunk])
try:
req = urllib.request.urlopen(url, timeout=10)
data = json.loads(req.read().decode('utf-8'))
if data.get('type') == 'multiinfo':
for item in data.get('results', []):
results[item['Name']] = item
except Exception:
pass
return results
def needs_upgrade(local_ver, aur_ver):
res = run_cmd(['vercmp', local_ver, aur_ver])
try:
return int(res.stdout.strip()) < 0
except Exception:
return False
def extract_ver(content):
pkgver, pkgrel = None, None
for line in content.split('\n'):
line = line.strip()
if line.startswith('pkgver='):
pkgver = line.split('=', 1)[1].strip('"\'')
elif line.startswith('pkgrel='):
pkgrel = line.split('=', 1)[1].strip('"\'')
return pkgver, pkgrel
def strip_volatile_fields(content):
content = re.sub(r'^pkgver=.*$', '', content, flags=re.MULTILINE)
content = re.sub(r'^pkgrel=.*$', '', content, flags=re.MULTILINE)
sum_patterns = [
r'md5sums', r'sha1sums', r'sha224sums', r'sha256sums',
r'sha384sums', r'sha512sums', r'b2sums', r'cksums'
]
for sp in sum_patterns:
pattern = r'^' + sp + r'(?:_[a-zA-Z0-9_]+)?=\(.*?\)'
content = re.sub(pattern, '', content, flags=re.MULTILINE | re.DOTALL)
lines = [line.rstrip() for line in content.split('\n') if line.strip()]
return '\n'.join(lines)
def main():
local_pkgs = get_local_foreign_packages()
if not local_pkgs:
sys.exit(0)
aur_info = get_aur_info(list(local_pkgs.keys()))
upgrades = []
for pkg, local_ver in local_pkgs.items():
if pkg in aur_info:
aur_ver = aur_info[pkg]['Version']
if needs_upgrade(local_ver, aur_ver):
upgrades.append((pkg, local_ver, aur_ver))
if not upgrades:
sys.exit(0)
print(f"Checking PKGBUILD diffs for {len(upgrades)} upgrades...")
temp_dir = tempfile.mkdtemp()
try:
for pkg, local_ver, aur_ver in upgrades:
print(f"Checking {pkg} ({local_ver} -> {aur_ver})")
repo_dir = os.path.join(temp_dir, pkg)
res = run_cmd(['git', 'clone', '--quiet', f'https://aur.archlinux.org/{pkg}.git', repo_dir])
if res.returncode != 0:
print(f"Failed to clone {pkg}", file=sys.stderr)
sys.exit(1)
res = run_cmd(['git', 'log', '--pretty=format:%H'], cwd=repo_dir)
commits = res.stdout.strip().split('\n')
old_pkgbuild_content = None
new_pkgbuild_content = None
with open(os.path.join(repo_dir, 'PKGBUILD'), 'r', encoding='utf-8', errors='ignore') as f:
new_pkgbuild_content = f.read()
local_ver_no_epoch = local_ver.split(':', 1)[-1]
parts = local_ver_no_epoch.rsplit('-', 1)
expected_ver, expected_rel = parts if len(parts) == 2 else (parts[0], "")
for commit in commits:
res = run_cmd(['git', 'show', f"{commit}:PKGBUILD"], cwd=repo_dir)
if res.returncode == 0:
content = res.stdout
p_ver, p_rel = extract_ver(content)
if p_ver == expected_ver and p_rel == expected_rel:
old_pkgbuild_content = content
break
if old_pkgbuild_content is None:
print(f"ERROR: Could not find old PKGBUILD for {pkg} version {local_ver}. Cannot safely verify diff.", file=sys.stderr)
sys.exit(1)
old_stripped = strip_volatile_fields(old_pkgbuild_content)
new_stripped = strip_volatile_fields(new_pkgbuild_content)
if old_stripped != new_stripped:
print(f"ERROR: Unsafe changes detected in PKGBUILD for {pkg}!", file=sys.stderr)
with open(os.path.join(temp_dir, 'old'), 'w') as f:
f.write(old_stripped)
with open(os.path.join(temp_dir, 'new'), 'w') as f:
f.write(new_stripped)
subprocess.run(['diff', '-u', os.path.join(temp_dir, 'old'), os.path.join(temp_dir, 'new')])
sys.exit(1)
finally:
shutil.rmtree(temp_dir)
if __name__ == '__main__':
main()