gitlab-cve-scanner.py

import json
import argparse
import requests
import socket
import re
import csv
from urllib.parse import urlparse
from datetime import datetime

import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class VulnersAPI:
    API_URL = "https://vulners.com/api/v3/burp/software/"
    HEADERS = {
        "User-Agent": "Vulners NMAP Plugin 1.7",
        "Accept-Encoding": "gzip, deflate"
    }

    @staticmethod
    def get_cves_by_version(version):
        if not version or version.lower().startswith("hash unknown"):
            return []

        params = {
            "software": "gitlab",
            "version": version,
            "type": "software"
        }

        try:
            response = requests.get(
                VulnersAPI.API_URL,
                headers=VulnersAPI.HEADERS,
                params=params,
                timeout=5
            )
            response.raise_for_status()
            data = response.json()

            if "data" not in data or "search" not in data["data"]:
                return []

            cves = []
            for vuln in data["data"]["search"]:
                if vuln["_source"]["type"] == "cve":
                    cves.append({
                        "id": vuln["_source"]["id"],
                        "cvss": vuln["_source"].get("cvss", {}).get("score", "N/A"),
                        "href": vuln["_source"].get("href", "N/A")
                    })

            return cves

        except requests.RequestException:
            print("[ERROR] Error requesting CVE.")
            return []

    @staticmethod
    def is_host_vulnerable(version, target_cve):
        cves = VulnersAPI.get_cves_by_version(version)
        return any(cve["id"] == target_cve for cve in cves)


class GitlabHashChecker:
    def __init__(self, hashes_file: str):
        with open(hashes_file, 'r', encoding='utf-8') as f:
            self.hash_map = json.load(f)

    def get_edition_and_version(self, full_hash: str):
        if full_hash not in self.hash_map:
            return ("unknown", "unknown")

        build_info = self.hash_map[full_hash]
        edition = build_info["edition"]
        versions = build_info["versions"]
        if not versions:
            return ("unknown", "unknown")

        version_str = versions[-1]
        return (edition, version_str)

    def find_unique_hash_by_prefix(self, prefix: str):
        matches = [h for h in self.hash_map if h.startswith(prefix)]
        return matches[0] if len(matches) == 1 else ""


class GitlabFetcher:
    def __init__(self, timeout: int = 5):
        self.timeout = timeout
        self.default_headers = {
            "User-Agent": (
                "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
                "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36"
            ),
            "Accept": (
                "text/html,application/xhtml+xml,application/xml;q=0.9,"
                "image/avif,image/webp,image/apng,*/*;q=0.8,"
                "application/signed-exchange;v=b3;q=0.7"
            )
        }

    def fetch_sign_in_revision(self, base_url: str) -> str:
        if base_url.endswith("/"):
            base_url = base_url[:-1]
        url = f"{base_url}/users/sign_in"

        try:
            resp = requests.get(url, timeout=self.timeout, verify=False, headers=self.default_headers)
            resp.raise_for_status()
            revision_regex = re.compile(r'\bgon\.revision\b\s*=\s*["\']?([0-9A-Za-z]+)["\']?', re.IGNORECASE)
            match = revision_regex.search(resp.text)
            if match:
                return match.group(1)
        except requests.RequestException:
            pass
        return ""

    def fetch_manifest_hash(self, base_url: str) -> str:
        if base_url.endswith("/"):
            base_url = base_url[:-1]
        url = f"{base_url}/assets/webpack/manifest.json"

        try:
            resp = requests.get(url, timeout=self.timeout, verify=False, headers=self.default_headers)
            resp.raise_for_status()
            data = resp.json()
            return data.get("hash", "")
        except (requests.RequestException, json.JSONDecodeError):
            pass
        return ""


class GitlabVersionDetector:
    def __init__(self, hashes_file: str, timeout: int = 5):
        self.hash_checker = GitlabHashChecker(hashes_file)
        self.fetcher = GitlabFetcher(timeout=timeout)

    def detect(self, raw_input: str):
        raw_input = raw_input.strip()
        has_scheme = raw_input.lower().startswith(("http://", "https://"))

        if has_scheme:
            final_url = raw_input
            return self._detect_from_base(final_url)
        else:
            for scheme_prefix in ["http://", "https://"]:
                final_url = scheme_prefix + raw_input
                result = self._detect_from_base(final_url)
                if result[0]:
                    return result
            return (False, final_url, "unknown", "unknown", "", "")

    def _detect_from_base(self, base_url: str):
        commit_hash_raw = self.fetcher.fetch_sign_in_revision(base_url)
        if commit_hash_raw:
            full_hash = self.hash_checker.find_unique_hash_by_prefix(commit_hash_raw)
            if full_hash:
                edition, version_str = self.hash_checker.get_edition_and_version(full_hash)
                return (True, base_url, edition, version_str, commit_hash_raw, "")
            else:
                return (True, base_url, "unknown", "unknown", commit_hash_raw, "")

        manifest_hash_raw = self.fetcher.fetch_manifest_hash(base_url)
        if manifest_hash_raw:
            edition, version_str = self.hash_checker.get_edition_and_version(manifest_hash_raw)
            return (True, base_url, edition, version_str, "", manifest_hash_raw)

        return (False, base_url, "unknown", "unknown", "", "")


def parse_host_and_port(raw_input: str):
    scheme_removed = re.sub(r'^https?://', '', raw_input.strip(), flags=re.IGNORECASE)
    parsed = urlparse(f"http://{scheme_removed}")
    host_part = parsed.netloc

    if ':' in host_part:
        host, port_str = host_part.split(':', 1)
    else:
        host, port_str = host_part, ""

    host = host.strip()
    port_str = port_str.strip()

    if not host:
        return "-", "-", ""

    try:
        socket.inet_aton(host)
        return ("-", host, port_str)
    except OSError:
        domain = host
        try:
            ip_ = socket.gethostbyname(domain)
        except Exception:
            ip_ = "-"
        return (domain, ip_, port_str)


def get_csv_filename():
    now_str = datetime.now().strftime("%Y%m%d_%H%M%S")
    return f"gitlab_report_{now_str}.csv"


def main():
    parser = argparse.ArgumentParser(description="Detect gitlab versions and find CVEs for these versions")
    parser.add_argument('-u', '--input_url', help='Single URL/host', default=None)
    parser.add_argument('-l', '--input_list', help='File with a list of URLs/hosts', default=None)
    parser.add_argument('-f', '--hash-file', help='Path to hashes.json', default='automation/hashes.json')
    parser.add_argument('-r', '--report', help='Write output to CSV file', action='store_true')
    parser.add_argument('--show-cves', help='Show CVEs for found versions', action='store_true')
    parser.add_argument('--cve', help='Check if vulnerable to specific CVE (example, CVE-2025-0555)', default=None)
    args = parser.parse_args()

    detector = GitlabVersionDetector(hashes_file=args.hash_file)
    results = []

    def process_line(line: str):
        line_clean = line.strip()
        if not line_clean:
            return

        success, final_url, edition, ver_str, commit_hash, manifest_hash = detector.detect(line_clean)
        domain_, ip_, user_port_str = parse_host_and_port(line_clean)

        parsed_final = urlparse(final_url)
        final_scheme = parsed_final.scheme
        final_host = parsed_final.hostname or ""
        final_port = parsed_final.port
        if final_port is None:
            final_port = 443 if final_scheme == "https" else 80
        url_with_port = f"{final_scheme}://{final_host}:{final_port}"
        actual_port_str = user_port_str if user_port_str else str(final_port)

        if edition != "unknown":
            console_msg = f"{url_with_port} - {edition} {ver_str}"
        else:
            parts = []
            if commit_hash:
                parts.append(f"commit hash '{commit_hash}'")
            if manifest_hash:
                parts.append(f"manifest hash '{manifest_hash}'")
            if parts:
                console_msg = f"{url_with_port} - {', '.join(parts)}"
            else:
                console_msg = f"{url_with_port} - failed to find hashes"

        if args.cve and edition != "unknown":
            vulnerable = VulnersAPI.is_host_vulnerable(ver_str, args.cve)
            if vulnerable:
                console_msg += " - Vulnerable"
            else:
                console_msg += " - Not vulnerable"

        print(console_msg)

        # Show CVEs
        cve_list = []
        if args.show_cves and edition != "unknown":
            cves = VulnersAPI.get_cves_by_version(ver_str)
            cve_list = [cve["id"] for cve in cves]
            cve_display = ", ".join(cve_list) if cve_list else "No known CVEs"
            print(f"CVEs: {cve_display}\n")

        row = {
            "Company": "",
            "URL": url_with_port,
            "Domain": domain_,
            "IP": ip_,
            "Port": actual_port_str,
            "Protocol": "tcp",
            "Edition": edition,
            "Version": ver_str
        }

        # Single CVE
        if args.cve:
            row["Name"] = args.cve
            if edition != "unknown":
                row["Is vulnerable"] = "Vulnerable" if vulnerable else "Not vulnerable"
            else:
                row["Is vulnerable"] = "Unknown (no version)"

        # Show CVEs in report
        if args.show_cves:
            row["CVEs"] = ", ".join(cve_list) if cve_list else "No known CVEs"

        results.append(row)

    if args.input_url:
        process_line(args.input_url)
    elif args.input_list:
        try:
            with open(args.input_list, 'r', encoding='utf-8') as f:
                for line in f:
                    process_line(line)
        except FileNotFoundError:
            print(f"File {args.input_list} not found.")
            return
    else:
        print("Example usage:\n"
              "  python3 main.py -u mydomain.com\n"
              "  python3 main.py -l input.txt\n"
              "  python3 main.py -l input.txt --report\n"
              "  python3 main.py -u mydomain.com --cve \"CVE-2025-0555\"\n"
              "  python3 main.py -u mydomain.com --cve \"CVE-2025-0555\" --report\n"
              "  python3 main.py -l input.txt --show-cves\n"
              "  python3 main.py -l input.txt --show-cves --report\n")
        return

    # Если нужно записать CSV
    if args.report and results:
        csv_file = get_csv_filename()

        fieldnames = ["Company", "URL", "Domain", "IP", "Port", "Protocol", "Edition", "Version"]

        if args.cve:
            fieldnames.append("Name")
            fieldnames.append("Is vulnerable")

        if args.show_cves:
            fieldnames.append("CVEs")

        with open(csv_file, mode='w', encoding='utf-8', newline='') as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames, delimiter=';')
            writer.writeheader()
            for row in results:
                writer.writerow(row)

        print(f"Report saved to file: {csv_file}")


if __name__ == "__main__":
    main()