utility.py

"""provide common utility attributes such as root path."""

import hashlib
import hmac
from datetime import datetime, timedelta
from functools import wraps
from ipaddress import IPv4Address, IPv6Address, ip_address, ip_network
from os import path
from typing import Callable, List, Union

import requests
import werkzeug
from flask import abort, g, redirect, request, send_file
from google.auth.exceptions import GoogleAuthError

ROOT_DIR = path.dirname(path.abspath(__file__))


def serve_file_download(file_name, file_folder, file_sub_folder='') -> werkzeug.wrappers.response.Response:
    """
    Serve file download by redirecting using Signed Download URLs.

    Falls back to the copy in the sample repository when the service account
    does not authenticate, which is the case in the Docker development stack:
    there the credentials are generated locally and ``/repository`` is a plain
    volume rather than a gcsfuse mount backed by a bucket.

    :param file_name: name of the file
    :type file_name: str
    :param file_folder: name of the folder
    :type file_folder: str
    :param file_sub_folder: sub folder of the original folder
    :type file_sub_folder: str
    :return: response, the file download
    :rtype: werkzeug.wrappers.response.Responsee
    """
    from run import config, storage_client_bucket

    file_path = path.join(file_folder, file_sub_folder, file_name)

    try:
        blob = storage_client_bucket.blob(file_path)
        blob.content_disposition = f'attachment; filename="{file_name}"'
        blob.patch()
        url = blob.generate_signed_url(
            version="v4",
            expiration=timedelta(minutes=config.get('GCS_SIGNED_URL_EXPIRY_LIMIT', '')),
            method="GET",
        )
        return redirect(url)
    except GoogleAuthError:
        # Only raised when the token exchange itself fails, so a working
        # service account never lands here and a Cloud Storage outage still
        # surfaces in production instead of being served a stale local copy.
        local_path = path.join(config.get('SAMPLE_REPOSITORY', ''), file_path)
        if path.isfile(local_path):
            return send_file(local_path, as_attachment=True, download_name=file_name)
        return abort(404)


def request_from_github(abort_code: int = 418) -> Callable:
    """Provide decorator to handle request from GitHub on the web hook."""
    def decorator(f):
        """Decorate the function to check if a request is a GitHub hook request."""
        @wraps(f)
        def decorated_function(*args, **kwargs):
            if request.method != 'POST':
                g.log.info(f"Request type {request.method} detected")

            request_ip = ip_address(f"{request.remote_addr}")
            if not is_github_web_hook_ip(request_ip):
                g.log.warning(f"Unauthorized attempt by IP {request_ip}")
                abort(abort_code)

            for header in ['X-GitHub-Event', 'X-GitHub-Delivery', 'X-Hub-Signature-256', 'User-Agent']:
                if header not in request.headers:
                    g.log.critical(f"{header} not in headers!")
                    abort(abort_code)

            ua = request.headers.get('User-Agent', '')
            if not ua.startswith('GitHub-Hookshot/'):
                g.log.critical("User-Agent does not begin with GitHub-Hookshot/!")
                abort(abort_code)

            if not request.is_json:
                g.log.critical("Request is not JSON!")
                abort(abort_code)

            return f(*args, **kwargs)

        return decorated_function

    return decorator


cached_web_hook_blocks: List[str] = []
cached_load_time: datetime = datetime(1970, 1, 1)


def cache_has_expired(load_time: datetime) -> bool:
    """
    Check if the cache expired.

    :return: True if the cache was last updated more than one hour ago.
    :rtype: bool
    """
    return load_time + timedelta(hours=1) < datetime.now()


def is_github_web_hook_ip(request_ip: Union[IPv4Address, IPv6Address]) -> bool:
    """
    Check if the given IP address is matching one provided by the API of GitHub.

    :param request_ip: The IP address the request came from.
    :type request_ip: IPAddress
    :return: True if the IP address is a valid GitHub Web Hook requester.
    :rtype: bool
    """
    for block in get_cached_web_hook_blocks():
        if request_ip in ip_network(block):
            return True

    return False


def get_cached_web_hook_blocks() -> List[str]:
    """
    Fetch the cached web hook blocks.

    :return: A list of ip blocks.
    :rtype: List[str]
    """
    global cached_web_hook_blocks
    global cached_load_time
    from run import config

    if len(cached_web_hook_blocks) == 0 or cache_has_expired(cached_load_time):
        client_id = config.get('GITHUB_CLIENT_ID', '')
        client_secret = config.get('GITHUB_CLIENT_KEY', '')
        meta_json = requests.get(
            'https://api.github.com/meta', auth=(client_id, client_secret)).json()
        try:
            cached_web_hook_blocks = meta_json['hooks']
            # We successfully fetched the IPs so we reset the clock
            cached_load_time = datetime.now()
        except KeyError:
            g.log.critical(f"Failed to retrieve hook IP's from GitHub! API returned {meta_json}")

    return cached_web_hook_blocks


def is_valid_signature(x_hub_signature, data, private_key):
    """
    Re-check if the GitHub hook request got valid signature.

    :param x_hub_signature: Value of the ``X-Hub-Signature-256`` header,
        e.g. ``sha256=<hex digest>``
    :type x_hub_signature: str
    :param data: Signature's data
    :type data: bytearray or bytes or str
    :param private_key: Signature's token
    :type private_key: str or bytes
    :return: False if the signature is missing, malformed, not SHA-256,
        or does not match.
    :rtype: bool
    """
    if not x_hub_signature or not isinstance(x_hub_signature, str):
        return False

    hash_algorithm, separator, github_signature = x_hub_signature.partition('=')
    if not separator or hash_algorithm != 'sha256':
        return False

    if private_key is None:
        return False
    if isinstance(private_key, str):
        encoded_key = private_key.encode('latin-1')
    elif isinstance(private_key, (bytes, bytearray)):
        encoded_key = bytes(private_key)
    else:
        return False

    if data is None:
        return False
    if isinstance(data, str):
        data = data.encode('utf-8')
    elif not isinstance(data, (bytes, bytearray)):
        return False

    try:
        mac = hmac.new(encoded_key, msg=data, digestmod=hashlib.sha256)
        # Compare bytes: compare_digest raises TypeError on non-ASCII str input.
        return hmac.compare_digest(mac.hexdigest().encode(), github_signature.encode('utf-8', 'replace'))
    except (ValueError, TypeError):
        return False