diff --git a/backend/gitea.py b/backend/gitea.py new file mode 100644 index 0000000..83c0859 --- /dev/null +++ b/backend/gitea.py @@ -0,0 +1,152 @@ +"""Fixed dedicated identity backend; no caller-selected hosts or admin proxy.""" +import base64 +import json +import os +import re +import urllib.error +import urllib.parse +import urllib.request + +ORIGIN = os.environ.get('PERSIMMONS_GITEA_ORIGIN', 'https://teahub.xyz').rstrip('/') +ALLOW_REGISTRATION = os.environ.get('PERSIMMONS_GITEA_ALLOW_REGISTRATION', '1') == '1' + + +class Failure(Exception): + def __init__(self, code=503): + self.code = code + super().__init__('Gitea request unavailable') + + +class NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, *args, **kwargs): + return None + + +def call(path, method='GET', body=None, login=None): + if login: + auth = 'Basic ' + base64.b64encode((login[0] + ':' + login[1]).encode()).decode() + else: + token = os.environ.get('PERSIMMONS_GITEA_TOKEN', '') + if not token: + raise Failure() + auth = 'token ' + token + request = urllib.request.Request(ORIGIN + '/api/v1' + path, method=method, + headers={'Authorization': auth, 'Accept': 'application/json', 'Content-Type': 'application/json'}, + data=json.dumps(body).encode() if body is not None else None) + try: + with urllib.request.build_opener(NoRedirect()).open(request, timeout=15) as response: + raw = response.read() + return json.loads(raw) if raw else None + except urllib.error.HTTPError as error: + raise Failure(error.code) from None + except (OSError, ValueError): + raise Failure() from None + + +def identity(user): + if (not isinstance(user, dict) or type(user.get('id')) is not int or user['id'] < 1 or + user.get('is_admin') is not False or not isinstance(user.get('login'), str) or + not re.fullmatch(r'[a-zA-Z0-9][a-zA-Z0-9_.-]{0,63}', user['login']) or + user.get('active') is False or user.get('prohibit_login') is True): + raise Failure(403) + return {'id': user['id'], 'login': user['login'], 'owner': 'pg-' + str(user['id'])} + + +def authenticate(name, password, register): + if register and not ALLOW_REGISTRATION: + raise Failure(403) + if register: + try: + created = call('/admin/users', 'POST', { + 'username': name, 'password': password, + 'email': name + '@users.persimmons.app', 'visibility': 'private', + 'must_change_password': False, 'send_notify': False, + 'restricted': False}) + except Failure as error: + # Gitea reports duplicate usernames/emails with 422. + raise Failure(409 if error.code in (409, 422) else error.code) from None + identity(created) # Fail closed if the server ever creates an admin. + # Basic auth is sent only over HTTPS to the pinned dedicated host, never retained. + user = identity(call('/user', login=(name, password))) + if user['login'].lower() != name.lower(): + raise Failure(403) + if register and user['id'] != created['id']: + raise Failure(403) + return user + + +def current(user): + found = identity(call('/users/' + urllib.parse.quote(user['login'], safe=''))) + if found['id'] != user['id'] or found['login'] != user['login']: + raise Failure(401) + return found + + +def repo_name(value): + if not isinstance(value, str) or not 1 <= len(value) <= 100: + raise ValueError() + name = re.sub(r'\s+', '-', value.strip()) + if (not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.-]{0,99}', name) or + name.lower().endswith(('.git', '.wiki'))): + raise ValueError() + return name + + +def checked_repo(repo, user, name, empty=False, repo_id=None): + if (not isinstance(repo, dict) or type(repo.get('id')) is not int or + repo.get('owner', {}).get('id') != user['id'] or + repo.get('owner', {}).get('login') != user['login'] or + repo.get('name') != name or repo.get('private') is not True or + (empty and repo.get('empty') is not True) or + (repo_id is not None and repo.get('id') != repo_id)): + raise Failure(502) + return {'id': repo['id'], 'name': repo['name'], 'owner': user['login'], + 'private': True, 'empty': repo.get('empty') is True, + 'url': ORIGIN + '/' + urllib.parse.quote(user['login'], safe='') + '/' + urllib.parse.quote(name, safe='')} + + +def create_repo(user, name): + current(user) + try: + repo = call('/admin/users/' + urllib.parse.quote(user['login'], safe='') + '/repos', 'POST', + {'name': name, 'private': True, 'auto_init': False, 'template': False}) + except Failure as error: + raise Failure(409 if error.code in (409, 422) else error.code) from None + checked_repo(repo, user, name, empty=True) + # Verify the persisted resource independently before recording it as a vault. + return checked_repo(call('/repos/' + user['login'] + '/' + name), user, name, + empty=True, repo_id=repo['id']) + + +def list_repos(user, rows): + current(user) + return [checked_repo(call('/repos/' + user['login'] + '/' + name), user, name, repo_id=repo_id) + for repo_id, name in rows] + + +def actions(user, vault, kind, name=None, method='GET', body=None): + """Dedicated, identity-checked Actions storage. Secrets have no read helper.""" + if kind not in ('secrets', 'variables') or (kind == 'secrets' and method not in ('PUT', 'DELETE')): + raise ValueError() + if name is not None and not re.fullmatch(r'PERSIM_JUICE_[0-9a-f]{32}', name): + raise ValueError() + current(user) + base = '/repos/' + user['login'] + '/' + vault['name'] + checked_repo(call(base), user, vault['name'], repo_id=vault['id']) + path = base + '/actions/' + kind + if name: + result = call(path + '/' + name.upper(), method, body) + else: + result = [] + for page in range(1, 11): + checked_repo(call(base), user, vault['name'], repo_id=vault['id']) + rows = call(path + '?limit=100&page=' + str(page)) + if not isinstance(rows, list): + raise Failure(502) + result.extend(rows) + if len(rows) < 100: + break + else: + raise Failure(409) # Never silently omit cleanup records. + checked_repo(call(base), user, vault['name'], repo_id=vault['id']) + return result