269 lines
13 KiB
Python
269 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""Archive Gitea releases. Standard library only; never executes source content."""
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import re
|
||
import tempfile
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
import zipfile
|
||
|
||
MAX_BYTES = 512 * 1024 * 1024
|
||
MARKER = '<!-- release-sync: git.bunnyruihan.com/ruihan/yidaconnector -->'
|
||
|
||
|
||
class SyncError(Exception):
|
||
pass
|
||
|
||
|
||
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
||
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
||
return None
|
||
|
||
|
||
class API:
|
||
def __init__(self, base, repo, token):
|
||
self.base, self.repo, self.token = base.rstrip('/'), repo, token
|
||
self.opener = urllib.request.build_opener(NoRedirect)
|
||
|
||
def request(self, method, path, body=None, binary=False, content_type=None):
|
||
url = self.base + '/api/v1/repos/' + self.repo + path
|
||
payload = body if binary else (json.dumps(body).encode() if body is not None else None)
|
||
headers = {'Authorization': 'token ' + self.token, 'Accept': 'application/json'}
|
||
if payload is not None:
|
||
headers['Content-Type'] = content_type or 'application/json'
|
||
for attempt in range(3):
|
||
try:
|
||
with self.opener.open(urllib.request.Request(url, data=payload, headers=headers, method=method), timeout=60) as response:
|
||
data = response.read()
|
||
return json.loads(data) if data else None
|
||
except urllib.error.HTTPError as e:
|
||
if e.code == 404 and method == 'GET':
|
||
return None
|
||
if method == 'GET' and e.code in (429, 500, 502, 503, 504) and attempt < 2:
|
||
time.sleep(2 ** attempt)
|
||
continue
|
||
raise SyncError(f'{method} {path.split("?")[0]} HTTP {e.code}') from None
|
||
except (urllib.error.URLError, TimeoutError, OSError):
|
||
if method == 'GET' and attempt < 2:
|
||
time.sleep(2 ** attempt)
|
||
continue
|
||
raise SyncError(f'{method} request failed; retry the workflow to reconcile state') from None
|
||
|
||
def pages(self, path):
|
||
result = []
|
||
for page in range(1, 10001):
|
||
data = self.request('GET', path + ('&' if '?' in path else '?') + f'limit=50&page={page}')
|
||
if not isinstance(data, list):
|
||
raise SyncError('Expected paginated list: ' + path)
|
||
if not data:
|
||
return result
|
||
result.extend(data)
|
||
raise SyncError('Pagination limit exceeded')
|
||
|
||
def download(self, asset, destination):
|
||
url = asset['browser_download_url']
|
||
origin = urllib.parse.urlsplit(self.base)
|
||
for attempt in range(3):
|
||
current = url
|
||
try:
|
||
for _ in range(5):
|
||
parsed = urllib.parse.urlsplit(current)
|
||
if parsed.scheme != 'https' or parsed.netloc != origin.netloc or parsed.username:
|
||
raise SyncError('Attachment URL or redirect is outside the approved HTTPS origin')
|
||
req = urllib.request.Request(current, headers={'Authorization': 'token ' + self.token})
|
||
try:
|
||
with self.opener.open(req, timeout=60) as response, open(destination, 'wb') as out:
|
||
count = 0
|
||
while True:
|
||
chunk = response.read(1024 * 1024)
|
||
if not chunk:
|
||
break
|
||
count += len(chunk)
|
||
if count > MAX_BYTES:
|
||
raise SyncError('Attachment exceeds 512 MiB limit')
|
||
out.write(chunk)
|
||
validate_file(destination, asset)
|
||
return digest(destination)
|
||
except urllib.error.HTTPError as e:
|
||
if e.code in (301, 302, 303, 307, 308) and e.headers.get('Location'):
|
||
current = urllib.parse.urljoin(current, e.headers['Location'])
|
||
continue
|
||
if e.code in (401, 403, 404):
|
||
raise SyncError(f'Attachment download HTTP {e.code}') from None
|
||
raise
|
||
raise SyncError('Too many attachment redirects')
|
||
except (urllib.error.URLError, TimeoutError, OSError):
|
||
if attempt == 2:
|
||
raise SyncError('Attachment download failed after retries') from None
|
||
time.sleep(2 ** attempt)
|
||
|
||
def list_assets(self, release_id):
|
||
# Gitea's assets endpoint is NOT paginated; page/limit are ignored.
|
||
data = self.request('GET', f'/releases/{release_id}/assets')
|
||
if not isinstance(data, list):
|
||
raise SyncError('Expected attachment list')
|
||
return data
|
||
|
||
|
||
def digest(path):
|
||
h = hashlib.sha256()
|
||
with open(path, 'rb') as f:
|
||
for block in iter(lambda: f.read(1024 * 1024), b''):
|
||
h.update(block)
|
||
return h.hexdigest()
|
||
|
||
|
||
def validate_file(path, asset):
|
||
if Path(path).stat().st_size != asset['size']:
|
||
raise SyncError('Attachment size mismatch: ' + asset['name'])
|
||
if asset['name'].lower().endswith('.zip'):
|
||
try:
|
||
with zipfile.ZipFile(path) as z:
|
||
if sum(i.file_size for i in z.infolist()) > 2 * 1024 ** 3:
|
||
raise SyncError('ZIP expanded size exceeds verification limit')
|
||
if z.testzip() is not None:
|
||
raise SyncError('ZIP CRC mismatch')
|
||
except (zipfile.BadZipFile, RuntimeError, NotImplementedError):
|
||
raise SyncError('Invalid or unsupported ZIP: ' + asset['name']) from None
|
||
|
||
|
||
def validate_names(assets):
|
||
names = [a['name'] for a in assets]
|
||
if len(names) != len(set(names)):
|
||
raise SyncError('Source has duplicate attachment names')
|
||
for a in assets:
|
||
if not a['name'] or '/' in a['name'] or '\\' in a['name'] or a['name'].startswith(('.sync-', '.backup-')):
|
||
raise SyncError('Unsafe or reserved attachment name')
|
||
if a['size'] < 0 or a['size'] > MAX_BYTES:
|
||
raise SyncError('Attachment size outside configured limit')
|
||
|
||
|
||
def release_body(source, source_url, hashes, source_sha):
|
||
notes = source.get('body') or ''
|
||
return notes + '\n\n---\n' + MARKER + '\n' + (
|
||
'本仓库仅归档发布附件；自动生成的 Source Code 压缩包不是原项目源码。\n\n'
|
||
f'来源：{source_url}\n\n源标签对应提交：`{source_sha}`\n\n'
|
||
f'原站发布时间：{source.get("published_at") or source.get("created_at") or "unknown"}\n\n'
|
||
'已校验附件 SHA-256：\n\n```text\n' +
|
||
'\n'.join(h + ' ' + name for name, h in sorted(hashes.items())) + '\n```\n'
|
||
)
|
||
|
||
|
||
def sync_one(src, dst, source, anchor, dry_run, directory):
|
||
tag = source['tag_name']
|
||
key = urllib.parse.quote(tag, safe='')
|
||
source_assets = src.list_assets(source['id'])
|
||
validate_names(source_assets)
|
||
source_tag = src.request('GET', '/tags/' + key)
|
||
if not source_tag or not source_tag.get('commit', {}).get('sha'):
|
||
raise SyncError('Source tag commit cannot be resolved')
|
||
source_sha = source_tag['commit']['sha']
|
||
hashes, files = {}, {}
|
||
for a in source_assets:
|
||
p = Path(directory) / ('source-' + str(a['id']))
|
||
hashes[a['name']] = src.download(a, p)
|
||
files[a['name']] = p
|
||
fresh = src.list_assets(source['id'])
|
||
identity = lambda rows: sorted((a['id'], a['name'], a['size']) for a in rows)
|
||
if identity(fresh) != identity(source_assets):
|
||
raise SyncError('Source attachment list changed during download; retry later')
|
||
target = dst.request('GET', '/releases/tags/' + key)
|
||
source_url = src.base + '/' + src.repo + '/releases/tag/' + key
|
||
body = release_body(source, source_url, hashes, source_sha)
|
||
desired = {'name': source.get('name') or tag, 'body': body, 'prerelease': source.get('prerelease', False)}
|
||
if target and MARKER not in (target.get('body') or ''):
|
||
raise SyncError('Target release exists but is not managed by this synchronizer')
|
||
result = {'tag': tag, 'source_sha': source_sha, 'created': target is None, 'assets': [], 'metadata_changed': False}
|
||
if target is None and not dry_run:
|
||
target = dst.request('POST', '/releases', dict(desired, tag_name=tag, target_commitish=anchor, draft=True))
|
||
assets = dst.list_assets(target['id']) if target else []
|
||
for source_asset in source_assets:
|
||
name, sha = source_asset['name'], hashes[source_asset['name']]
|
||
existing = next((a for a in assets if a['name'] == name), None)
|
||
old_sha = None
|
||
if existing:
|
||
old_sha = dst.download(existing, Path(directory) / ('target-' + str(existing['id'])))
|
||
if old_sha == sha:
|
||
result['assets'].append({'name': name, 'sha256': sha, 'action': 'unchanged'})
|
||
continue
|
||
action = 'replace' if existing else 'add'
|
||
if not dry_run:
|
||
rid = target['id']
|
||
# Recover an upload accepted by the server before a previous connection failed.
|
||
temp_name = '.sync-' + hashlib.sha256((name + sha).encode()).hexdigest()[:24]
|
||
staging = next((a for a in assets if a['name'] == temp_name), None)
|
||
if staging is None:
|
||
staging = dst.request('POST', f'/releases/{rid}/assets?name=' + urllib.parse.quote(temp_name, safe=''),
|
||
files[name].read_bytes(), binary=True, content_type='application/octet-stream')
|
||
assets.append(staging)
|
||
staging_sha = dst.download(staging, Path(directory) / ('staged-' + str(staging['id'])))
|
||
if staging_sha != sha:
|
||
raise SyncError('Uploaded attachment SHA-256 mismatch')
|
||
if existing:
|
||
backup_name = '.backup-' + str(existing['id']) + '-' + name
|
||
dst.request('PATCH', f'/releases/{rid}/assets/{existing["id"]}', {'name': backup_name})
|
||
existing['name'] = backup_name
|
||
dst.request('PATCH', f'/releases/{rid}/assets/{staging["id"]}', {'name': name})
|
||
staging['name'] = name
|
||
result['assets'].append({'name': name, 'sha256': sha, 'action': action})
|
||
if target:
|
||
changed = target.get('draft', False) or any(target.get(k) != v for k, v in desired.items())
|
||
result['metadata_changed'] = changed
|
||
if changed and not dry_run:
|
||
dst.request('PATCH', f'/releases/{target["id"]}', dict(desired, draft=False))
|
||
else:
|
||
result['metadata_changed'] = True
|
||
return result
|
||
|
||
|
||
def main():
|
||
p = argparse.ArgumentParser()
|
||
p.add_argument('--dry-run', action='store_true')
|
||
p.add_argument('--tag', default='')
|
||
p.add_argument('--verify-all', action='store_true', help='All attachments are always fully verified')
|
||
p.add_argument('--report', default='sync-report.json')
|
||
args = p.parse_args()
|
||
report = {'dry_run': args.dry_run, 'started_at': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()), 'releases': [], 'errors': []}
|
||
try:
|
||
src = API('https://git.bunnyruihan.com:16666', 'ruihan/yidaconnector', os.environ['SOURCE_GITEA_TOKEN'])
|
||
dst = API('https://git.jskdjs.cn', 'ruihan/yidaconnector', os.environ['TARGET_GITEA_TOKEN'])
|
||
anchor = os.environ['ARCHIVE_ANCHOR_SHA']
|
||
if not re.fullmatch('[0-9a-f]{40}', anchor):
|
||
raise SyncError('Invalid archive anchor SHA')
|
||
info = dst.request('GET', '')
|
||
if not info or info['full_name'] != dst.repo or not info['private']:
|
||
raise SyncError('Expected private target archive repository')
|
||
releases = [r for r in src.pages('/releases') if not r.get('draft')]
|
||
if args.tag:
|
||
releases = [r for r in releases if r['tag_name'] == args.tag]
|
||
if not releases:
|
||
raise SyncError('Requested published source tag not found')
|
||
for source in reversed(releases):
|
||
try:
|
||
with tempfile.TemporaryDirectory(prefix='release-sync-') as directory:
|
||
result = sync_one(src, dst, source, anchor, args.dry_run, directory)
|
||
report['releases'].append(result)
|
||
print(json.dumps(result, ensure_ascii=False), flush=True)
|
||
except Exception as e:
|
||
# Do not render untrusted response bodies, request headers or tokens.
|
||
message = str(e) if isinstance(e, SyncError) else type(e).__name__
|
||
report['errors'].append({'tag': source['tag_name'], 'error': message})
|
||
print('FAILED ' + source['tag_name'] + ': ' + message, flush=True)
|
||
except Exception as e:
|
||
report['errors'].append({'error': str(e) if isinstance(e, SyncError) else type(e).__name__})
|
||
finally:
|
||
report['finished_at'] = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
|
||
Path(args.report).write_text(json.dumps(report, ensure_ascii=False, indent=2))
|
||
print(f'SUMMARY releases={len(report["releases"])} errors={len(report["errors"])} dry_run={args.dry_run}')
|
||
return 1 if report['errors'] else 0
|
||
|
||
|
||
if __name__ == '__main__':
|
||
raise SystemExit(main())
|