koji-forgejo-webhook/koji_forgejo_webhook.py

import json
import hashlib
import hmac
import logging
import configparser
import subprocess
from flask import Flask, request, abort, jsonify
import koji

# Load configuration from .ini file
config = configparser.ConfigParser()
config.read('/opt/koji-forgejo-webhook/config.ini')

app = Flask(__name__, instance_relative_config=True)
# app.config.from_mapping(config['DEFAULT'])
app.config['DEBUG'] = True
app.config['SECRET_KEY'] = config.get('DEFAULT', 'SECRET_KEY')
app.config['KOJI_SERVER'] = config.get('DEFAULT', 'KOJI_SERVER')
app.config['KOJI_TARGET'] = config.get('DEFAULT', 'KOJI_TARGET')
app.config['KOJI_USE_SHELL'] = config.getboolean('DEFAULT', 'KOJI_USE_SHELL')
app.config['KOJI_NOWAIT'] = config.getboolean('DEFAULT', 'KOJI_NOWAIT')
app.config['KOJI_SCRATCH'] = config.getboolean('DEFAULT', 'KOJI_SCRATCH')
app.config['WEBHOOK_SECRET_KEY'] = config.get('DEFAULT', 'WEBHOOK_SECRET_KEY')


class KojiProcessor:
    def __init__(self):
        pass

    def koji_build_shell(self, build_target: str, git_url: str,
                         scratch: bool = True, nowait: bool = True) -> str:
        result = {"out": "", "err": ""}

        # Construct the koji build command
        nw = '--nowait' if nowait else ''
        sc = '--scratch' if scratch else ''
        command = ['sudo', 'koji', 'build', sc, nw, build_target, git_url]
        try:
            # Start the build process
            exec = subprocess.run(' '.join(command), shell=True, capture_output=True, text=True)
            result['err'] = exec.stderr
            result['out'] = exec.stdout

            # Print output from the command
            logging.info(f"Build started successfully!")
            logging.info(exec.stdout)
        # except subprocess.CalledProcessError as e:
        #     logging.error("Error starting build:")
        except Exception as e:
            result['err'] = f"{e}"

        return result

    def koji_build(self, server_url, build_target, source_url, nowait=True):
        """
        Perform a Koji build
        
        :param server_url: Koji hub server URL
        :param build_target: Target for the build (e.g., 'f40-candidate')
        :param source_url: Git repository URL to build from
        """
        # Create a Koji client session
        session = koji.ClientSession(server_url)
        
        try:
            # Authenticate (if required)
            # Uncomment if authentication is needed
            session.login()
            
            # Submit the build task
            build_id = session.build(
                source_url,  # Source URL
                build_target,  # Build target
                opts={
                    'scratch': False,  # Set to True for scratch build
                    'nowait': nowait   # Wait for build completion
                }
            )
            return build_id
        
        except Exception as e:
            print(f"Build failed: {e}")
            return None


class ForgejoWebhookProcessor:
    def __init__(self, secret_key):
        self.secret_key = secret_key

    def validate_webhook(self, payload, signature):
        """Validate webhook signature"""
        computed_signature = hmac.new(
            self.secret_key.encode('utf-8'),
            payload.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()

        return hmac.compare_digest(computed_signature, signature)

    def process_webhook(self, payload):
        """Process webhook payload"""
        result = dict()
        try:
            data = json.loads(payload)

            # Extract key webhook information
            repository = data.get('repository', {})
            commits = data.get('commits', [])
            head_commit = data.get('head_commit')
            ref = data.get('ref', '')

            # Log basic webhook details
            logging.info(f"Repository: {repository.get('full_name')}")
            logging.info(f"Branch: {ref}")
            logging.info(f"Number of Commits: {len(commits)}")

            # result = head_commit.get("id")
            # Custom processing logic here
            # for commit in commits:
            #     logging.info(f"Commit: {commit.get('id')}")
            #     logging.info(f"Message: {commit.get('message')}")

            commit_id = head_commit.get("id")
            clone_url = repository.get("clone_url")

            git_url = f"git+{clone_url}#{commit_id}"

            scratch = True
            if 'release' in ref.lower():
                scratch = False

            if app.config.get('KOJI_USE_SHELL'):
                result = KojiProcessor().koji_build_shell(app.config['KOJI_TARGET'], git_url, scratch=scratch)
            else:
                result = KojiProcessor().koji_build(app.config['KOJI_SERVER'], app.config['KOJI_TARGET'], git_url)

        except json.JSONDecodeError:
            logging.error("Invalid JSON payload")
            abort(400)
        return result


@app.route('/ping', methods=['GET'])
def ping():
    result = subprocess.run('sudo koji moshimoshi', shell=True, capture_output=True, text=True)
    stdout = result.stdout
    stderr = result.stderr
    return jsonify({
        "out": f"{stdout}",
        "err": f"{stderr}",
    })


@app.route('/forgejo', methods=['POST'])
def webhook():
    processor = ForgejoWebhookProcessor(app.config.get('WEBHOOK_SECRET_KEY'))

    # Get payload and signature
    payload = request.get_data(as_text=True)
    signature = request.headers.get('X-Forgejo-Signature', '')

    # Validate webhook
    if not processor.validate_webhook(payload, signature):
        abort(403)

    # Process webhook
    return jsonify(processor.process_webhook(payload))


if __name__ == '__main__':
    logging.basicConfig(filename='/opt/koji-forgejo-webhook/error.log', level=logging.DEBUG)
    app.run(host="0.0.0.0", port=5001)