app/backend/gitea.py

"""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