#!/usr/bin/env python
"""
Script to identify packages for which the source code repository does not exists anymore in the SCM.

The objective is to prevent (defensive) or leverage (offesnive) the attack described in the article below:

    https://sockpuppets.medium.com/how-i-hacked-ctx-and-phpass-modules-656638c6ec5e
    BACKUP PDF: https://github.com/righettod/toolbox-pentest-web/issues/19#issuecomment-1138208780

Dependencies:
    pip install requests termcolor tabulate tqdm

Commands to list all dependencies depending on the package manager used:
    npm list --all > source.txt
    gradlew -q dependencies > source.txt
    mvn dependency:list > source.txt
    pipreqs --force --savepath source.txt .

For PHP, it is more straightforward thanks to the fact that "composer" provide directly the source code repository URL for a package:
    1. Obtain the collection of source code repository URL for every installed package:
        composer show -f json | jq -r ".installed[].source"
    2. Verify that each returned URL lead to an HTTP 200:
        for u in $(composer show -f json | jq -r ".installed[].source"); do echo -n "[+] $u: "; curl -skI $u | head -1 | cut -d' ' -f2 ; done

References:
    https://github.com/bndr/pipreqs
    https://pypi.org/project/pipreqs/
    https://stedolan.github.io/jq/
"""
import requests
import re
import argparse
import time
from urllib.parse import urlparse
from termcolor import colored
from tqdm import tqdm
from tabulate import tabulate


# Configuration
NPM_REGISTRY = "https://registry.npmjs.org/[PKG_NAME]/[PKG_VERSION]"
MAVEN_AND_GRADLE_REGISTRY = "https://repo1.maven.org/maven2/[GROUP_ID]/[ARTIFACT_ID]/[VERSION]/[ARTIFACT_ID]-[VERSION].pom"
PIPY_REGISTRY = "https://pypi.python.org/pypi/[PKG_NAME]/[PKG_VERSION]/json"
NPM_EXTRACTION_REGEX = r'([@\/a-zA-Z0-9\-_\.]+@[a-zA-Z0-9\-_\.]+)'
MAVEN_EXTRACTION_REGEX = r'([a-zA-Z0-9\-_\.]+:[a-zA-Z0-9\-_\.]+:[a-z]+:[a-zA-Z0-9\-_\.]+)'
GRADLE_EXTRACTION_REGEX = r'---\s([a-zA-Z0-9\-_\.]+:[a-zA-Z0-9\-_\.]+:[a-zA-Z0-9\-_\.]+)'
PIPY_EXTRACTION_REGEX = r'([a-zA-Z0-9\-_\.]+==[a-zA-Z0-9\-_\.]+)'
HTTP_REQ_TIMEOUT_IN_SECONDS = 20
HTTP_429_MAX_WAITING_ROUND = 20
HTTP_429_MAX_WAITING_TIME_IN_SECONDS = 10
HTTP_REDIRECT_CODES = [301, 302, 303, 307, 308]

# Utility functions


def extract_package_infos(package_full_identifier, package_manager):
    if package_manager == "npm":
        # As package name can also include a '@' then I cannot split on this character
        last_arobase_position = package_full_identifier.rfind("@")
        infos = {"PKG_NAME": package_full_identifier[:last_arobase_position], "PKG_VERSION": package_full_identifier[last_arobase_position+1:]}
    elif package_manager == "maven":
        parts = package_full_identifier.split(":")
        infos = {"GROUP_ID": parts[0], "ARTIFACT_ID": parts[1], "VERSION": parts[3]}
    elif package_manager == "gradle":
        parts = package_full_identifier.split(":")
        infos = {"GROUP_ID": parts[0], "ARTIFACT_ID": parts[1], "VERSION": parts[2]}
    else:
        parts = package_full_identifier.split("==")
        infos = {"PKG_NAME": parts[0], "PKG_VERSION": parts[1]}
    return infos


def extract_packages_list(file_location, package_manager):
    already_processed = []
    with open(file_location, mode="r", encoding="utf-8") as f:
        text_content = f.read()
    packages_list = []
    if package_manager == "npm":
        regex = NPM_EXTRACTION_REGEX
    elif package_manager == "maven":
        regex = MAVEN_EXTRACTION_REGEX
    elif package_manager == "gradle":
        regex = GRADLE_EXTRACTION_REGEX
    else:
        regex = PIPY_EXTRACTION_REGEX
    packages = re.findall(regex, text_content, re.MULTILINE)
    for package in packages:
        if package not in already_processed:
            packages_list.append(extract_package_infos(package, package_manager))
            already_processed.append(package)
    return packages_list


def extract_package_code_source_repo_url(package_infos, package_manager, req_session):
    repo_url = None
    if package_manager == "npm":
        npm_source_url_attributes = ["repository", "bugs", "author"]
        url = NPM_REGISTRY.replace("[PKG_NAME]", package_infos["PKG_NAME"]).replace("[PKG_VERSION]", package_infos["PKG_VERSION"])
        try:
            response = req_session.get(url, timeout=HTTP_REQ_TIMEOUT_IN_SECONDS)
            if response.status_code == 200:
                data = response.json()
                for npm_source_url_attribute in npm_source_url_attributes:
                    if npm_source_url_attribute in data and "url" in data[npm_source_url_attribute]:
                        repo_url = data[npm_source_url_attribute]["url"]
                        if npm_source_url_attribute == "bug":
                            repo_url = repo_url[:repo_url.rfind("/issues")]
                        break
        except Exception as e:
            repo_url = f"[ERROR]: {str(e)}"
    elif package_manager in ["gradle", "maven"]:
        gid_path = package_infos["GROUP_ID"].replace(".", "/")
        url = MAVEN_AND_GRADLE_REGISTRY.replace("[GROUP_ID]", gid_path).replace("[ARTIFACT_ID]", package_infos["ARTIFACT_ID"]).replace("[VERSION]", package_infos["VERSION"])
        try:
            response = req_session.get(url, timeout=HTTP_REQ_TIMEOUT_IN_SECONDS)
            body = response.text
            if response.status_code == 200 and "<scm>" in body:
                tag_start = body.find("<scm>")
                tag_end = body.find("</scm>")
                body = body[tag_start:tag_end]
                if "<url>" in body:
                    tag_start = body.find("<url>")
                    tag_end = body.find("</url>")
                    body = body[tag_start:tag_end]
                    repo_url = body[5:]
        except Exception as e:
            repo_url = f"[ERROR]: {str(e)}"
    else:
        url = PIPY_REGISTRY.replace("[PKG_NAME]", package_infos["PKG_NAME"]).replace("[PKG_VERSION]", package_infos["PKG_VERSION"])
        try:
            response = req_session.get(url, timeout=HTTP_REQ_TIMEOUT_IN_SECONDS)
            if response.status_code == 200:
                metadata = response.json()
                if metadata["info"]["project_urls"] is not None and "Source" in metadata["info"]["project_urls"]:
                    repo_url = metadata["info"]["project_urls"]["Source"]
                elif metadata["info"]["project_urls"] is not None and "Homepage" in metadata["info"]["project_urls"]:
                    repo_url = metadata["info"]["project_urls"]["Homepage"]
        except Exception as e:
            repo_url = f"[ERROR]: {str(e)}"
    return repo_url


def convert_repo_url_to_http(repo_url):
    url = repo_url

    if url.startswith("scm:git:git@"):
        url = url.replace("scm:git:git@", "https://")
    elif url.startswith("git+ssh://git@"):
        url = url.replace("git+ssh://git@", "https://")
    elif url.startswith("git+"):
        url = url[4:]
    elif url.startswith("git://"):
        url = url.replace("git://", "https://")
    elif url.startswith("https://www.github.com/"):
        url = url.replace("https://www.github.com/", "https://github.com/")
    elif url.startswith("github.com/"):
        url = f"https://{url}"
    elif url.startswith("git@github.com:"):
        url = url.replace("git@github.com:", "https://github.com/")
    elif url.startswith("github.com:"):
        url = url.replace("github.com:", "https://github.com/")

    if url.endswith(".git"):
        url = url[:url.rfind(".git")]

    return url


def check_code_source_repo_online_existence(repo_url, req_session):
    is_valid = False
    waiting_round_count = 0
    http_response_code = -1
    http_redirection_target = ""
    try:
        response = req_session.head(repo_url, timeout=HTTP_REQ_TIMEOUT_IN_SECONDS, allow_redirects=False)
        while response.status_code in [429, 500] and waiting_round_count < HTTP_429_MAX_WAITING_ROUND:
            time.sleep(HTTP_429_MAX_WAITING_TIME_IN_SECONDS)
            response = req_session.head(repo_url, timeout=HTTP_REQ_TIMEOUT_IN_SECONDS, allow_redirects=False)
            waiting_round_count += 1
        is_valid = (response.status_code == 200)
        http_response_code = response.status_code
        if response.status_code in HTTP_REDIRECT_CODES:
            http_redirection_target = response.headers["Location"]
    except Exception:
        http_response_code = -1
        is_valid = False
        http_redirection_target = ""
    return (is_valid, http_response_code, http_redirection_target)


def create_package_diplay_identifier(package_infos):
    return ":".join(package_infos.values())


def shorten_redirection_url(repo_url, redirection_url):
    repo = urlparse(repo_url)
    redir = urlparse(redirection_url)
    if repo.scheme == redir.scheme and repo.netloc == redir.netloc:
        return redir.path
    else:
        return redirection_url


# Entry point
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Script to identify packages for which the source code repository does not exists anymore in the SCM.")
    required_params = parser.add_argument_group("required named arguments")
    required_params.add_argument("-i", action="store", dest="source_file", type=str, help="Path to the source text file containing the raw output of the dependencies listing command.", required=True)
    required_params.add_argument("-p", action="store", dest="package_manager", help="Type of package manager used.", choices=["npm", "gradle", "maven", "pipy"], required=True)
    parser.add_argument("-s", action="store_true", dest="skip_redirect", default=False, help="Consider source code repository redirection as a valid source code repository.", required=False)
    args = parser.parse_args()
    req_session = requests.Session()
    file_location = args.source_file
    package_manager = args.package_manager
    skip_redirect = args.skip_redirect
    print(colored(f"[+] Extract the list of packages from the source file '{file_location}'...", "yellow"))
    packages = extract_packages_list(file_location, package_manager)
    print(f"{len(packages)} packages identified.")
    print(colored(f"[+] Extract, for each package, its source code repository url via the package manager '{package_manager.upper()}' online metadata...", "yellow"))
    packages_src_repo_urls = {}
    packages_with_none_or_error = []
    packages_with_none_or_error.append(["Package", "Source code repo status"])
    for i in tqdm(range(len(packages))):
        package = packages[i]
        repo_url = extract_package_code_source_repo_url(package, package_manager, req_session)
        display_identifier = create_package_diplay_identifier(package)
        if repo_url is None:
            packages_with_none_or_error.append([display_identifier, "Not found."])
        elif repo_url.startswith("[ERROR]"):
            packages_with_none_or_error.append([display_identifier, repo_url])
        else:
            packages_src_repo_urls[display_identifier] = repo_url
    if len(packages_with_none_or_error) > 1:
        print(tabulate(packages_with_none_or_error, headers="firstrow", tablefmt="github"))
    print(colored(f"[+] For package with a identified source code repository url, check that it exists (exclude redirection enabled: {skip_redirect})...", "yellow"))
    packages_with_non_existing_repo = []
    packages_with_non_existing_repo.append(["Package", "Source code repo URL", "HTTP response code"])
    for package_identifier, repo_url in tqdm(packages_src_repo_urls.items()):
        http_url = convert_repo_url_to_http(repo_url)
        is_reachable, http_response_code, http_redirection_target = check_code_source_repo_online_existence(http_url, req_session)
        if not is_reachable:
            if http_response_code in HTTP_REDIRECT_CODES and skip_redirect:
                continue
            http_detail = http_response_code
            if len(http_redirection_target) > 0:
                # Remove common prefix between the repo url and the redirection url to shorten the display
                target_redirect = shorten_redirection_url(http_url, http_redirection_target)
                http_detail = f"{http_detail} ({target_redirect})"
            packages_with_non_existing_repo.append([package_identifier, http_url, http_detail])
    if len(packages_with_non_existing_repo) > 1:
        print(tabulate(packages_with_non_existing_repo, headers="firstrow", tablefmt="github"))
    else:
        print("[" + colored(f"V", "green") + "] All packages with a identified source code repository are reacheable.")
