copr.py

""" WSGI-based  filtering reverse proxy for github push webhooks to COPR. """

import json
import re
import traceback
from urllib.parse import parse_qs, unquote
from uuid import UUID
import hashlib
import hmac
import os
from http.client import responses
import requests
from yaml import safe_load

class ProxyError(BaseException):
    """ A simple error class. """

    def __init__(self, code, reason='', debug=False):
        self.code = code
        self.reason = reason
        self.debug = debug

    def __str__(self):
        """ Return standard code and statustext """
        return f'{self.code} {responses[self.code]}'

    def statustext(self):
        """ Return statustext """
        return responses[self.code]

class CoprProxy:
    """ A simple reverse proxy for COPR using WSGI. """

    def __init__(self, env, start_response):
        """ init our environment. """
        self.env = env
        self.start_response = start_response
        self.err = env['wsgi.errors']
        config = env.get('config')
        if config is None:
            raise ProxyError(503, 'config not specified')
        try:
            with open(config, 'r', encoding='utf-8') as f:
                self.cfg  = safe_load(f)
        except (ValueError, OSError) as ex:
            raise ProxyError(503, f'Unable to read config from {config}') from ex
        pcfgdir = self.cfg.get('projectcfgdir')
        if pcfgdir is not None:
            try:
                if os.path.realpath(pcfgdir, strict=True) != pcfgdir:
                    raise ProxyError(503, 'Invalid projectcfgdir', self._debug())
            except OSError as ex:
                raise ProxyError(503, 'Unable to use projectcfgdir', self._debug()) from ex
            try:
                found = False
                with os.scandir(pcfgdir) as it:
                    for e in it:
                        if not e.name.startswith('.') and e.is_file():
                            found = True
                if not found:
                    delattr(self.cfg, 'projectcfgdir')
            except OSError as ex:
                delattr(self.cfg, 'projectcfgdir')
                raise ProxyError(503, 'Unable to use projectcfgdir', self._debug()) from ex

    def _loadproj(self, proj, uuid):
        """ Load project config, if it exists """
        pcfgdir = self.cfg.get('projectcfgdir')
        if pcfgdir is not None:
            projcfg = os.path.join(pcfgdir, f'{proj}-{uuid}.yaml')
            try:
                if os.path.realpath(projcfg, strict=True) != projcfg:
                    raise ProxyError(503, 'Invalid proj or uuid', self._debug())
            except OSError as ex:
                raise ProxyError(503, 'Invalid proj or uuid', self._debug()) from ex
            if os.path.isfile(projcfg):
                try:
                    with open(projcfg, 'r', encoding='utf-8') as f:
                        pcfg  = safe_load(f)
                    for key in ['secret', 'paths', 'branches', 'tags']:
                        old = self.cfg.get(key)
                        self.cfg[key] = pcfg.get(key, old)
                except (ValueError, OSError) as ex:
                    raise ProxyError(503, 'Unable to read project config', self._debug()) from ex

    def _debug(self):
        """ Return the configured debug flag """
        ret = self.cfg.get('debug')
        if ret is None:
            return False
        return ret

    def _dryrun(self):
        """ Return the configured dryrun flag """
        ret = self.cfg.get('dryrun')
        if ret is None:
            return False
        return ret

    def _strict(self):
        """ Return the configured strict flag """
        ret = self.cfg.get('strict')
        if ret is None:
            return True
        return ret

    def _secret(self):
        """ Return the configured secret """
        ret = self.cfg.get('secret')
        if ret is None and self._strict():
            raise ProxyError(503, 'Strict validation enabled, but no secret configured locally',
                             self._debug())
        return ret

    def _select(self, what):
        """ Return the configured items """
        ret = self.cfg.get(what)
        if ret is None:
            return ret
        if isinstance(ret, str):
            return [ret]
        if isinstance(ret, (list, tuple)):
            return ret
        raise ProxyError(503, f'Invalid type of {what}. Must be a str, a list or a tuple',
                         self._debug())

    def _paths(self):
        """ Return the configured paths """
        return self._select('paths')

    def _branches(self):
        """ Return the configured branches """
        return self._select('branches')

    def _tags(self):
        """ Return the configured tags """
        return self._select('tags')

    def _proxies(self):
        """ fetch proxies from environment. """
        ret = {}
        for scheme in ['http', 'https']:
            value = self.env.get(f'{scheme}_proxy')
            if value is not None:
                ret[scheme] = value
        return ret

    def _checkuuid(self, uuid):
        """ Validate UUID. """
        if uuid is None:
            raise ProxyError(400, 'Missing uuid', self._debug())
        try:
            tuuid = str(UUID(uuid)).lower()
        except ValueError as ex:
            raise ProxyError(400, 'Invalid UUID', self._debug()) from ex
        if tuuid != uuid.lower():
            raise ProxyError(400, 'Invalid UUID', self._debug())

    def _mpmatch(self, patterns, s):
        """ Match multiple patterns on a string.  """
        if patterns is None or not patterns:
            return True
        if s is None:
            return False
        for pat in patterns:
            cre = re.compile(glob_to_re(pat))
            if cre.match(s):
                return True
        return False

    def _lmatch(self, pattern, slist):
        """ Match pattern on a list of strings.  """
        if slist is None or not slist:
            return False
        cre = re.compile(glob_to_re(pattern))
        for s in slist:
            if cre.match(s):
                return True
        return False

    def _sigvalidate(self):
        """Verify that the payload was sent from GitHub by validating SHA256.

        Signature is provided by github in a Header like this:

        X-Hub-Signature-256: sha256=757107ea0eb2509fc211221cce984b8a37570b6d7586c22c46f4379c8b043e17

        Returns body, if secret is None or has been validated successfully,

        See: https://docs.github.com/en/webhooks/webhook-events-and-payloads#delivery-headers
        """
        body = self.env['wsgi.input'].read(self._contentlen()).decode('utf-8')
        secret = self._secret()
        if secret is None:
            return body
        sig = self.env.get('HTTP_X_HUB_SIGNATURE_256')
        if sig is None:
            raise ProxyError(403, 'Signature missing', self._debug())
        hobj = hmac.new(secret.encode('utf-8'), msg=body.encode('utf-8'), digestmod=hashlib.sha256)
        expected = 'sha256=' + hobj.hexdigest()
        if hmac.compare_digest(expected, sig):
            return body
        raise ProxyError(403, 'Signature validation failed', self._debug())

    def _contentlen(self):
        """ Get content length """
        clen = self.env.get('CONTENT_LENGTH', '0')
        if not re.match(r'[0-9]+$', clen):
            raise ProxyError(400, 'Invalid Content-Length', self._debug())
        if int(clen) > 20971520:
            raise ProxyError(413, 'Invalid Content-Length', self._debug())
        return int(clen)

    def _urlparams(self):
        """ Handle URL parameters """
        try:
            qparams =  parse_qs(unquote(self.env['QUERY_STRING']), strict_parsing=True,
                                max_num_fields=3)
            proj = qparams.get('proj')
            if proj is None:
                raise ProxyError(400, 'Missing or empty proj', self._debug())
            proj = proj[0]
            if not re.match('^[0-9]+$', proj):
                raise ProxyError(400, 'Invalid proj', self._debug())
            uuid = qparams.get('uuid')
            if uuid is not None:
                uuid = uuid[0]
            self._checkuuid(uuid)
            # Now that we have proj and uuid, we can load a project-specifig config,
            # if it exists.
            self._loadproj(proj, uuid)
            pkg = qparams.get('pkg')
            if pkg is not None:
                pkg = pkg[0]
        except ValueError as ex:
            raise ProxyError(400, 'Too many query parameters', self._debug()) from ex

        return {'proj': proj, 'uuid': uuid, 'pkg': pkg}

    def _branchandtagmatch(self, obj):
        """ Handle tag and branch matching """
        tags = self._tags()
        branches = self._branches()
        if tags is None and branches is None:
            return True
        tag = None
        branch = None
        if 'ref' in obj:
            ref = obj['ref']
            if ref.startswith('refs/heads/'):
                branch = re.sub(r'^refs/heads/', '', ref)
            elif ref.startswith('refs/tags/'):
                tag = re.sub(r'^refs/tags/', '', ref)
                ref = obj['base_ref']
                if ref.startswith('refs/heads/'):
                    branch = re.sub(r'^refs/heads/', '', ref)
            return self._mpmatch(tags, tag) and self._mpmatch(branches, branch)
        raise ProxyError(400, 'missing ref', self._debug())

    def _pathmatch(self, obj):
        """ Handle path matching """
        paths = self._paths()
        if paths is None:
            return True
        if 'commits' in obj:
            candidates = []
            for c in obj['commits']:
                candidates += c['added'] + c['modified'] + c['removed']
            if 'head_commit' in obj:
                c = obj['head_commit']
                candidates += c['added'] + c['modified'] + c['removed']
            for pat in paths:
                if self._lmatch(pat, candidates):
                    return True
        else:
            raise ProxyError(400, 'missing commits', self._debug())
        return False

    def forward(self, dst, ua, ctype, body):
        """ Forward request to destination. """
        hdrs = {}
        hdrs['User-Agent'] = ua
        hdrs['Content-Type'] = ctype
        hdrs['Content-Length'] = self.env['CONTENT_LENGTH']
        for key in self.env.keys():
            m = re.search(r'^HTTP_X_GITHUB_(\S+)', key)
            if m is not None and m.group(1):
                hk = '-'.join(word.capitalize() for word in m.group(1).split('_'))
                hdrs[f'X-Github-{hk}'] = self.env[key]
            m = re.search(r'^HTTP_X_HUB_(\S+)', key)
            if m is not None and m.group(1):
                hk = '-'.join(word.capitalize() for word in m.group(1).split('_'))
                hdrs[f'X-Hub-{hk}'] = self.env[key]
        try:
            return requests.post(dst, headers=hdrs, data=body, proxies=self._proxies(),
                              timeout=10)
        except requests.Timeout as ex:
            raise ProxyError(504, ex.args[0], self._debug()) from ex
        except (requests.RequestException, requests.TooManyRedirects,
                requests.JSONDecodeError) as ex:
            raise ProxyError(500, ex.args[0], self._debug()) from ex

    def formatdst(self, projectid, uuid, pkgname):
        """ Format destination URL. """
        try:
            if pkgname is None:
                fmt = self.cfg.get('copr_url_2')
                if fmt is None:
                    raise ProxyError(503, 'Missing URL template copr_url_nopkg', self._debug())
                return fmt.format(projectid=projectid, uuid=uuid)
            fmt = self.cfg.get('copr_url_3')
            if fmt is None:
                raise ProxyError(503, 'Missing URL template copr_url_pkg', self._debug())
            return fmt.format(projectid=projectid, uuid=uuid, pkgname=pkgname)
        except KeyError as ex:
            raise ProxyError(503, f'Misconfigured url template. Missing key: {ex}',
                             self._debug()) from ex

    def handle(self):
        """ Handle one request. """

        ua = self.env.get('HTTP_USER_AGENT')
        ctype = self.env.get('CONTENT_TYPE')
        ghe  = self.env.get('HTTP_X_GITHUB_EVENT')

        # Basic sanity checks
        if ua is None or not re.match(r'GitHub-Hookshot/.+', ua):
            raise ProxyError(403, 'User-Agent does not match GitHub-Hookshot/', self._debug())
        if ctype is None or not ctype == 'application/json':
            raise ProxyError(403, 'Invalid Content-Type', self._debug())
        if ghe is None:
            raise ProxyError(403, 'Missing X-GitHub-Event', self._debug())

        if self.env['REQUEST_METHOD'] == 'POST':

            if ghe != 'push':
                raise ProxyError(400, 'Not a push event', self._debug())

            q = self._urlparams()
            body = self._sigvalidate()

            try:
                obj = json.loads(body)
            except (json.JSONDecodeError, UnicodeDecodeError) as ex:
                raise ProxyError(400, 'Invalid JSON', self._debug()) from ex

            mbat = self._branchandtagmatch(obj)
            mpat = self._pathmatch(obj)
            if mbat and mpat:
                dst = self.formatdst(q['proj'], q['uuid'], q['pkg'])
                if self._dryrun():
                    print(f'Found pattern in commit, would forward to {dst}', file=self.err)
                else:
                    print(f'Found pattern in commit, forwarding to {dst}', file=self.err)
                    r = self.forward(dst, ua, ctype, body)
                    self.start_response(f'{r.status_code} {r.reason}', [])
                    return [r.content]
            else:
                if self._debug():
                    pretty = json.dumps(obj, indent=2)
                    print(f'Unmatched mbat={mbat} mpat={mpat} body=\n{pretty}', file=self.err)

        self.start_response('200 OK', [])
        return []

def application(env, start_response):
    """ WSGI entrypoint. """
    try:
        rp = CoprProxy(env, start_response)
        return rp.handle()
    except ProxyError as ex:
        print(f'{ex.code}: {ex.reason}', file=env['wsgi.errors'])
        if ex.debug:
            traceback.print_exception(ex, file=env['wsgi.errors'])
        start_response(str(ex), [('Content-Type', 'text/plain; charset=utf-8')])
        if ex.code in [403, 503]:
            # For security reasons, do not expose the cause.
            return []
        # For other errors, expose the cause.
        return [str(ex.reason).encode('utf-8')]

def glob_to_re(pat: str) -> str:
    # pylint: disable=locally-disabled, too-many-nested-blocks, too-many-branches
    """Translate a shell PATTERN to a regular expression modified to provide ** matching

    Based on https://stackoverflow.com/a/72400344/5030772
    Posted by Mathew Wicks, modified by community. See post 'Timeline' for change history
    Retrieved 2025-12-27, License - CC BY-SA 4.0

    Derived from `fnmatch.translate()` of Python version 3.8.13
    SOURCE: https://github.com/python/cpython/blob/v3.8.13/Lib/fnmatch.py#L74-L128
    """

    i, n = 0, len(pat)
    res = ''
    while i < n:
        c = pat[i]
        i = i+1
        if c == '*':
            # -------- CHANGE START --------
            # prevent '*' matching directory boundaries, but allow '**' to match them
            j = i
            if j < n and pat[j] == '*':
                res = res + '.*'
                i = j+1
            else:
                res = res + '[^/]*'
            # -------- CHANGE END ----------
        elif c == '?':
            # -------- CHANGE START --------
            # prevent '?' matching directory boundaries
            res = res + '[^/]'
            # -------- CHANGE END ----------
        elif c == '[':
            j = i
            if j < n and pat[j] == '!':
                j = j+1
            if j < n and pat[j] == ']':
                j = j+1
            while j < n and pat[j] != ']':
                j = j+1
            if j >= n:
                res = res + '\\['
            else:
                stuff = pat[i:j]
                if '--' not in stuff:
                    stuff = stuff.replace('\\', r'\\')
                else:
                    chunks = []
                    k = i+2 if pat[i] == '!' else i+1
                    while True:
                        k = pat.find('-', k, j)
                        if k < 0:
                            break
                        chunks.append(pat[i:k])
                        i = k+1
                        k = k+3
                    chunks.append(pat[i:j])
                    # Escape backslashes and hyphens for set difference (--).
                    # Hyphens that create ranges shouldn't be escaped.
                    stuff = '-'.join(s.replace('\\', r'\\').replace('-', r'\-')
                                     for s in chunks)
                # Escape set operations (&&, ~~ and ||).
                stuff = re.sub(r'([&~|])', r'\\\1', stuff)
                i = j+1
                if stuff[0] == '!':
                    # -------- CHANGE START --------
                    # ensure sequence negations don't match directory boundaries
                    stuff = '^/' + stuff[1:]
                    # -------- CHANGE END ----------
                elif stuff[0] in ('^', '['):
                    stuff = '\\' + stuff
                res = f'{res}[{stuff}]'
        else:
            res = res + re.escape(c)
    return fr'(?s:{res})\Z'