path: root/src/main.py

# Copyright (C) 2024 The Qt Company Ltd.
# Contact: https://www.qt.io/licensing/
#
# You may use this file under the terms of the 3-clause BSD license.
# See the file LICENSE in qt/qtrepotools for details.
#

"""This script listens for incoming webhook requests of patchset-created type
    from Gerrit, checks out the patch locally, and checks security headers on it.
    It then posts a comment for each issue identified to Gerrit with the results.
"""

import asyncio
import base64
import fnmatch
import json
import logging
import os
import re
import shutil
import sys
import time
import traceback
from functools import wraps
from logging.handlers import TimedRotatingFileHandler
from subprocess import CalledProcessError

import aiohttp
from aiohttp import web

# Configure logging
LOG_DIR = "logging"
os.makedirs(LOG_DIR, exist_ok=True)
LOG_FILE = os.path.join(LOG_DIR, "qtsecuritybot.log")
handler = TimedRotatingFileHandler(LOG_FILE, when='midnight', backupCount=90)
handler.setFormatter(logging.Formatter(
    '%(asctime)s - %(levelname)s - %(message)s'))
logging.basicConfig(level=logging.INFO, handlers=[handler])
log = logging.getLogger()

GERRIT_USERNAME = os.environ.get('QTSECURITYBOT_GERRIT_USERNAME')
GERRIT_PASSWORD = os.environ.get('QTSECURITYBOT_GERRIT_PASSWORD')

if not GERRIT_USERNAME or not GERRIT_PASSWORD:
    log.info(
        'Please set the QTSECURITYBOT_GERRIT_USERNAME and QTSECURITYBOT_GERRIT_PASSWORD environment variables.')
    sys.exit(1)

# Base64 encode the username and password
GERRIT_AUTH = GERRIT_USERNAME + ':' + GERRIT_PASSWORD
GERRIT_AUTH = GERRIT_AUTH.encode('utf-8')
GERRIT_AUTH = base64.b64encode(GERRIT_AUTH).decode('utf-8')

CONFIG = {
    'CLONE_TIMEOUT': 240,
    'CHECKOUT_TIMEOUT': 60,
    'MAX_RETRIES': 10,
    'RETRY_DELAY': 5,
    'MAX_LOCK_WAIT': 900,  # 15 minutes,
    'TEAMS_URL': os.environ.get('QTSECURITYBOT_TEAMS_WEBHOOK_URL'),
    'TEAMS_ERROR_URL': os.environ.get('QTSECURITYBOT_TEAMS_ERROR_WEBHOOK_URL'),
}


class Lock:
    """A simple lock class to prevent multiple events from
       being handled at the same time.
    """

    def __init__(self):
        self.locked = False
        self._lock_time = 0

    async def acquire(self):
        """Acquire the lock with timeout."""
        start_time = time.time()
        while self.locked:
            if time.time() - start_time > CONFIG['MAX_LOCK_WAIT']:
                raise TimeoutError("Lock acquisition timed out")
            await asyncio.sleep(1)
        self.locked = True
        self._lock_time = time.time()

    def release(self):
        """Release the lock."""
        self.locked = False
        self._lock_time = 0


def log_errors(f):
    """Decorator to log any unhandled errors in a function."""
    @wraps(f)
    async def wrapper(*args, **kwargs):
        try:
            return await f(*args, **kwargs)
        except Exception as e:
            log.error("Error in %s: %s\n%s", f.__name__,
                      str(e), traceback.format_exc())
            raise
    return wrapper


semaphore = Lock()


async def clone_repo(data):
    """Clone the target repo and check out the branch if needed."""
    log.info("Cloning repo %s", data['change']['project'])
    repo_name = data['repo_name']

    if os.path.exists(repo_name):
        # return if the repo already exists
        return
    repo_url = "https://codereview.qt-project.org/" + \
        data['change']['project'] + ".git"

    if not shutil.which('git'):
        raise FileNotFoundError("Git executable not found in PATH")

    try:
        p = await asyncio.create_subprocess_exec(
            'git', 'clone', repo_url, repo_name,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        stdout, stderr = await asyncio.wait_for(p.communicate(), timeout=CONFIG['CLONE_TIMEOUT'])

        if p.returncode != 0:
            raise CalledProcessError(
                p.returncode, 'git clone', output=stdout, stderr=stderr)

        log.info("Repository cloned successfully: %s", stdout.decode())
    except FileNotFoundError as e:
        log.error("Error: %s", e)
    except asyncio.TimeoutError:
        log.error("Error: Subprocess timed out")
    except CalledProcessError as e:
        raise e


async def checkout_patch(data):
    """Check out the patch."""

    log.info("%s: Checking out patch", data['change']['number'])
    # Check out the patch
    repo_dir = data['repo_name']
    # git clean -fdx first to remove any untracked files
    p = await asyncio.create_subprocess_exec('git', '-C', repo_dir, 'clean', '-fdx')
    await p.communicate()
    try:
        # Fetch the patch first
        p = await asyncio.create_subprocess_exec(
            'git', '-C', repo_dir, 'fetch', 'origin', data['patchSet']['ref'],
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        stdout, stderr = await asyncio.wait_for(p.communicate(), timeout=240)
        if p.returncode != 0:
            raise CalledProcessError(
                p.returncode, 'git fetch', output=stdout, stderr=stderr)
        log.info("Fetch successful: %s", stdout.decode())

        p = await asyncio.create_subprocess_exec(
            'git', '-C', repo_dir, 'checkout', 'FETCH_HEAD',
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        stdout, stderr = await asyncio.wait_for(p.communicate(), timeout=240)
        if p.returncode != 0:
            raise CalledProcessError(
                p.returncode, 'git checkout', output=stdout, stderr=stderr)
        log.info("Checkout successful: %s", stdout.decode())

    except FileNotFoundError as e:
        log.error("Error: %s", e)
    except asyncio.TimeoutError:
        log.error("Error: Subprocess timed out")
    except CalledProcessError as e:
        raise e


async def run_security_header_check(data):
    """Run securityHeader on the patch."""
    log.info("%s: Running Security Header checks", data['change']['number'])
    comments_per_file = {}

    def _check_header_lines(lines, file_name, is_new_patch):
        line_number = None
        # Start line numbering at 1
        for _line_number, line in enumerate(lines, 1):
            if _line_number > 50:
                break  # Only check the first 50 lines

            # Check for security headers. Match lazily. Enforcement is done via sanity bot.
            comment_pattern = re.compile(
                r'^\s*(#|\/\/|\/\*|\*|<!--|--)\s*Qt-Security\s*score:\s*critical'
            )

            for _line_number, line in enumerate(lines):
                if comment_pattern.search(line):
                    line_number = _line_number
                    break

        if line_number:
            # Add the comment to the list of comments for this file
            comments_per_file[file_name] = {
                'line': line_number,
                'message': "Modifying security sensitive file.",
                'side': 'REVISION' if is_new_patch else 'PARENT'
            }
    # End of _check_header_lines

    # Get the list of files changed in this patch
    try:
        p = await asyncio.create_subprocess_exec(
            'git', '-C', data['repo_name'], 'diff-tree', '--no-commit-id',
            '--name-status', '-r', 'FETCH_HEAD',
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        stdout, stderr = await asyncio.wait_for(p.communicate(), timeout=240)
        if p.returncode != 0:
            raise CalledProcessError(
                p.returncode, 'git diff-tree', output=stdout, stderr=stderr)
        log.info("Diff successful: %s", stdout.decode())

    except FileNotFoundError as e:
        log.error("Error: %s", e)
    except asyncio.TimeoutError:
        log.error("Error: Subprocess timed out")
    except CalledProcessError as e:
        raise e
    # Parse the output
    changed_files = []  # Empty list to store file names
    for line in stdout.decode().split('\n'):
        if len(line) > 0 and not line.startswith("D"):  # Skip empty lines and deleted files
            # Add the file name to the list
            changed_files.append(line.split('\t')[1])

    ignore_patterns = []

    # Run securityHeader on each file
    for file in changed_files:
        if any(fnmatch.fnmatch(file, pattern) for pattern in ignore_patterns):
            continue  # Skip this file if it matches any of the other ignore patterns

        file_name = os.path.relpath(file, os.getcwd())

        # Get how the file was modified (added, modified, or deleted)
        try:
            p = await asyncio.create_subprocess_exec(
                'git', '-C', data['repo_name'], 'diff-tree', '--no-commit-id',
                '--name-status', '-r', 'FETCH_HEAD', '--', file,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            stdout, stderr = await asyncio.wait_for(p.communicate(), timeout=240)
            if p.returncode != 0:
                raise CalledProcessError(p.returncode, 'git diff-tree',
                                         output=stdout, stderr=stderr
                                         )
            log.info("File diff successful: %s", stdout.decode())

        except FileNotFoundError as e:
            log.error("Error: %s", e)
        except asyncio.TimeoutError:
            log.error("Error: Subprocess timed out")
        except CalledProcessError as e:
            raise e

        change_type = None
        for line in stdout.decode().split('\n'):
            if len(line) > 0:
                change_type = line.split('\t')[0]

        # Mofified files need to be checked for header in both the current and previous versions

        try:
            # For added files and modified files, Open the file and scan for security headers.
            if change_type == 'A' or change_type == 'M':
                with open(os.path.join(data['repo_name'], file_name), 'r', encoding='utf-8') as f:
                    lines = f.readlines(8192)
                    _check_header_lines(lines, file_name, True)

            # For deleted files or modified files, use the previous version of the file.
            if file_name not in comments_per_file and file_name in changed_files:
                if change_type == 'D' or change_type == 'M':
                    try:
                        p = await asyncio.create_subprocess_exec(
                            'git', '-C', data['repo_name'], 'show', 'FETCH_HEAD~1:' + file,
                            stdout=asyncio.subprocess.PIPE,
                            stderr=asyncio.subprocess.PIPE
                        )
                        stdout, stderr = await asyncio.wait_for(p.communicate(), timeout=240)
                        if p.returncode != 0:
                            raise CalledProcessError(p.returncode, 'git show',
                                                    output=stdout, stderr=stderr
                                                    )
                        log.debug("Show file successful: %s", stdout.decode())
                        del_file = stdout.decode()
                        lines = del_file.split('\n')
                        _check_header_lines(lines, file_name, False)
                    except FileNotFoundError as e:
                        log.error("Error: %s", e)
                    except asyncio.TimeoutError:
                        log.error("Error: Subprocess timed out")
                    except CalledProcessError as e:
                        raise e
        except UnicodeDecodeError as e:
            log.warning("WARN: File %s not valid utf-8 encoding", file_name)
        except IsADirectoryError as e:
            log.warning("WARN: File %s is a directory", file_name)

    if not comments_per_file:
        return None

    log.info("%s: Comments: %s", data['change']
             ['number'], json.dumps(comments_per_file))
    return comments_per_file


def generate_review(comments_per_file):
    """Generate a review from the comments."""

    log.info("Generating review")
    if not comments_per_file or len(comments_per_file.keys()) == 0:
        return None
    comment_inputs = {}
    for file_name, comment in comments_per_file.items():
        if file_name not in comment_inputs:
            comment_inputs[file_name] = []
            comment_inputs[file_name].append({
                'line': comment['line'],
                'message': comment['message'],
                'side': comment['side'],
                'unresolved': 'false'
            })
    review = {
        'comments': comment_inputs
    }
    return review


async def post_review(data, review, retry=0):
    """Post the review to Gerrit."""
    change_number = data['change']['number']
    revision = data['patchSet']['revision']
    url = f"https://codereview.qt-project.org/a/changes/{change_number}/revisions/{revision}/review"
    headers = {'Content-Type': 'application/json;charset=UTF-8',
               'Authorization': 'Basic ' + GERRIT_AUTH}  # Ensure GERRIT_AUTH is defined

    log.info("%s: Posting review", change_number)
    log.info('%s: Review data: %s', change_number, json.dumps(review))

    try:
        async with aiohttp.ClientSession() as session:
            async with session.post(url, json=review, headers=headers) as response:
                if response.status == 409 and retry < 10:
                    log.info(
                        '%s: Retrying due to 409 Lock Failure...', change_number)
                    await asyncio.sleep(5)
                    await post_review(data, review, retry + 1)
                elif response.status >= 400:
                    response_text = await response.text()
                    log.info('Error posting review: %s %s',
                             response.status, response_text)
                else:
                    log.info('%s: Review posted successfully.', change_number)
    except aiohttp.ClientError as e:
        log.info('Error posting review: %s', str(e))


async def post_hashtag(data, retry=0):
    """Post the review to Gerrit."""
    change_number = data['change']['number']
    url = f"https://codereview.qt-project.org/a/changes/{change_number}/hashtags"
    headers = {'Content-Type': 'application/json;charset=UTF-8',
               'Authorization': 'Basic ' + GERRIT_AUTH}
    hashtags = {"add": ["Qt-Security change"]}

    log.info('%s: Review data: %s', change_number, hashtags)
    try:
        async with aiohttp.ClientSession() as session:
            async with session.post(url, json=hashtags, headers=headers) as response:
                if response.status == 409 and retry < 10:
                    log.info(
                        '%s: Retrying due to 409 Lock Failure...', change_number)
                    await asyncio.sleep(5)
                    await post_hashtag(data, retry + 1)
                elif response.status >= 400:
                    response_text = await response.text()
                    log.info('Error posting review: %s %s',
                             response.status, response_text)
                else:
                    log.info('%s: Hashtag posted successfully.', change_number)
    except aiohttp.ClientError as e:
        log.info('Error posting hashtag: %s', str(e))


async def post_teams_message(data, retry=0):
    """Post a simple message to Teams with the change details."""
    if not CONFIG['TEAMS_URL']:
        log.info('Teams webhook URL not set. Skipping message posting.')
        return
    headers = {
        'Content-Type': 'application/json'
    }
    card = {
        "@type": "MessageCard",
        "@context": "http://schema.org/extensions",
        "summary": "New Security Relevant Change",
        "themeColor": "0076D7",
        "title": data['change']['subject'],
        "sections": [{
            "activityTitle": data['change']['subject'],
            "activitySubtitle": f"Change number: {data['change']['number']}",
            "facts": [{
                "name": "Project:",
                "value": data['change']['project']
            }, {
                "name": "Branch:",
                "value": data['change']['branch']
            }],
            "markdown": True
        }],
        "potentialAction": [{
            "@type": "OpenUri",
            "name": "View Change",
            "targets": [{
                "os": "default",
                "uri": data['change']['url']
            }]
        }]
    }

    try:
        async with aiohttp.ClientSession() as session:
            async with session.post(CONFIG['TEAMS_URL'], json=card, headers=headers) as response:
                if response.status == 408 and retry < 10:
                    log.info('%s: Retrying due to 408 Request Timeout...',
                             data['change']['number'])
                    await asyncio.sleep(5)
                    await post_teams_message(data, retry + 1)
                elif response.status >= 400:
                    response_text = await response.text()
                    log.info('Error posting Teams message: %s %s',
                             response.status, response_text)
                else:
                    log.info('%s: Teams message posted successfully.',
                             data['change']['number'])
    except aiohttp.ClientError as e:
        log.info('Error posting Teams message: %s', str(e))


async def post_teams_error_message(data, custom_text, retry=0):
    """Post an error message to Teams with the change details for diagnostic purposes."""
    if not CONFIG['TEAMS_ERROR_URL']:
        log.info('Teams error webhook URL not set. Skipping error message posting.')
        return
    headers = {
        'Content-Type': 'application/json'
    }
    card = {
        "@type": "MessageCard",
        "@context": "http://schema.org/extensions",
        "summary": "New Change",
        "themeColor": "0076D7",
        "title": data['change']['subject'],
        "text": custom_text,
        "sections": [{
            "activityTitle": data['change']['subject'],
            "activitySubtitle": f"Change number: {data['change']['number']}",
            "facts": [{
                "name": "Project:",
                "value": data['change']['project']
            }, {
                "name": "Branch:",
                "value": data['change']['branch']
            }],
            "markdown": True
        }],
        "potentialAction": [{
            "@type": "OpenUri",
            "name": "View Change",
            "targets": [{
                "os": "default",
                "uri": data['change']['url']
            }]
        }]
    }

    try:
        async with aiohttp.ClientSession() as session:
            async with session.post(CONFIG['TEAMS_ERROR_URL'], json=card, headers=headers) as res:
                if res.status == 408 and retry < 10:
                    log.info('%s: Retrying due to 408 Request Timeout...',
                             data['change']['number'])
                    await asyncio.sleep(5)
                    await post_teams_message(data, retry + 1)
                elif res.status >= 400:
                    response_text = await res.text()
                    log.info('Error posting Teams message: %s %s',
                             res.status, response_text)
                else:
                    log.info('%s: Teams message posted successfully.',
                             data['change']['number'])
    except aiohttp.ClientError as e:
        log.info('Error posting Teams message: %s', str(e))


@log_errors
async def handle(request):
    """Handle the incoming webhook request."""
    # Wrap the whole function in case the request is malformed.
    try:
        body = await request.text()
        data = json.loads(body)

        # Validate request
        required_fields = ['type', 'change', 'patchSet']
        if not all(field in data for field in required_fields):
            return web.Response(status=400, text="Missing required fields")

        # Make sure the change is in state NEW
        if data['change']['status'] != 'NEW':
            return web.Response(status=204)

        # make sure it's a patchset-created event
        if data['type'] != 'patchset-created':
            return web.Response(status=204)

        # Only act on pyside repos.
        if not data['change']['project'].startswith('pyside'):
            return web.Response(status=204)

        data['repo_name'] = data['change']['project'].split('/')[-1]

        log.info("%s: Received webhook for %s",
                 data['change']['number'], data['patchSet']['revision'])

        # Request a lock on the git repo
        try:
            log.debug("%s: Acquiring lock", data['change']['number'])
            await semaphore.acquire()
            await clone_repo(data)
            await checkout_patch(data)
            comments = await run_security_header_check(data)
        except Exception as e:
            log.error("Error: %s", str(e))
            await post_teams_error_message(data, str(e))
            return web.Response(status=500, body=str(e))
        finally:
            log.info("%s: Releasing lock", data['change']['number'])
            semaphore.release()

        # create a review with the comments if any python files were reviewed
        if comments:
            review = generate_review(comments)
            await post_review(data, review)
            await post_hashtag(data)
            # Post a message to Teams about the Security relevant change.
            # Re-enable this line to post to Teams about security Changes
            # It is currently disabled to prevent spamming users.
            # await post_teams_message(data)
        else:
            log.info("%s: Change not security relevant",
                     data['change']['number'])

        return web.Response(status=200)
    except json.JSONDecodeError:
        return web.Response(status=400, text="Invalid JSON")
    except Exception as e:
        await post_teams_error_message(data, f"{str(e)}\n{traceback.format_exc()}")
        return web.Response(status=500)


async def handle_status(req):
    """Handle the status request."""
    return web.Response(status=200)


async def run_web_server():
    """Run the web server."""
    app = web.Application()
    app.add_routes([web.get('/status', handle_status)])
    app.add_routes([web.post('/', handle)])
    runner = web.AppRunner(app)
    await runner.setup()
    port = os.environ.get("QTSECURITYBOT_PORT") or 8088
    site = web.TCPSite(runner, '0.0.0.0', port)
    await site.start()
    log.info("Web server started on port %s", port)


def main():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        loop.run_until_complete(run_web_server())
        loop.run_forever()
    except KeyboardInterrupt:
        pass
    finally:
        loop.close()


if __name__ == "__main__":
    main()