git clone https://github.com/icarax/gitea-flaw-mitigation.git
cd gitea-flaw-mitigation

# Create a virtual environment (recommended)
python -m venv venv
source venv/bin/activate   # Windows: venv\Scripts\activate

# Install dependencies
pip install --upgrade pip
pip install requests python-dotenv tqdm

# Initialize a new npm project (if you don't have one)
npm init -y

# Install core dependencies
npm install axios dotenv

# Install TypeScript and type definitions (dev deps)
npm install --save-dev typescript @types/node ts-node

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "commonjs",
    "strict": true,
    "esModuleInterop": true,
    "forceConsistentCasingInFileNames": true,
    "skipLibCheck": true
  }
}

#!/usr/bin/env python3
"""
detect_and_mitigate_gitea.py

Detects a vulnerable Gitea version and, if found, disables the
repository migration feature via the admin API.

Requirements:
    pip install requests python-dotenv tqdm
"""

import os
import sys
import logging
from typing import Tuple, Optional

import requests
from dotenv import load_dotenv
from tqdm import tqdm

# ----------------------------------------------------------------------
# Configuration & Logging
# ----------------------------------------------------------------------
load_dotenv()  # loads .env into os.environ

GITEA_URL = os.getenv("GITEA_URL", "").rstrip("/")
GITEA_TOKEN = os.getenv("GITEA_TOKEN", "")

if not GITEA_URL or not GITEA_TOKEN:
    sys.exit("❌ Error: GITEA_URL and GITEA_TOKEN must be set in .env")

HEADERS = {
    "Authorization": f"token {GITEA_TOKEN}",
    "Accept": "application/json",
}

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger(__name__)

# ----------------------------------------------------------------------
# Helper functions
# ----------------------------------------------------------------------
def get_json(url: str, params: dict = None) -> dict:
    """GET JSON with basic error handling."""
    try:
        resp = requests.get(url, headers=HEADERS, params=params, timeout=10)
        resp.raise_for_status()
        return resp.json()
    except requests.RequestException as exc:
        log.error("Failed to GET %s: %s", url, exc)
        raise


def patch_json(url: str, payload: dict) -> dict:
    """PATCH JSON with basic error handling."""
    try:
        resp = requests.patch(
            url, headers=HEADERS, json=payload, timeout=10
        )
        resp.raise_for_status()
        return resp.json()
    except requests.RequestException as exc:
        log.error("Failed to PATCH %s: %s", url, exc)
        raise


def fetch_version() -> Tuple[int, int, int]:
    """Return Gitea version as (major, minor, patch)."""
    data = get_json(f"{GITEA_URL}/api/v1/version")
    version_str = data.get("version", "0.0.0")
    # Strip possible pre-release suffix (e.g., "1.15.0-dev")
    version_str = version_str.split("-")[0]
    parts = [int(p) for p in version_str.split(".")]
    # Pad to 3 components
    while len(parts) < 3:
        parts.append(0)
    return tuple(parts)  # type: ignore


def is_vulnerable(version: Tuple[int, int, int]) -> bool:
    """Check against known vulnerable ranges."""
    major, minor, patch = version
    if major == 1:
        if minor == 13 and 0 <= patch <= 6:
            return True
        if minor == 14 and 0 <= patch <= 2:
            return True
        if minor == 15 and patch == 0:  # includes 1.15.0-dev treated as 0 patch
            return True
    return False


def get_migration_enabled() -> bool:
    """Return True if repository.migrate.enable is true."""
    settings = get_json(f"{GITEA_URL}/api/v1/admin/settings")
    # The setting is nested under "repository" -> "migrate" -> "enable"
    migrate = settings.get("repository", {}).get("migrate", {}).get("enable")
    if isinstance(migrate, bool):
        return migrate
    # Fallback: some versions expose as string "true"/"false"
    if isinstance(migrate, str):
        return migrate.lower() == "true"
    raise ValueError("Unable to determine migration setting type")


def disable_migration() -> None:
    """Set repository.migrate.enable => false."""
    payload = {"repository.migrate.enable": False}
    patch_json(f"{GITEA_URL}/api/v1/admin/settings", payload)
    log.info("🔒 Migration feature disabled.")


# ----------------------------------------------------------------------
# Main workflow
# ----------------------------------------------------------------------
def main() -> None:
    log.info("🔍 Checking Gitea instance at %s", GITEA_URL)

    # 1️⃣ Get version
    version = fetch_version()
    log.info("📦 Detected Gitea version: %s", ".".join(map(str, version)))

    # 2️⃣ Determine vulnerability
    vulnerable = is_vulnerable(version)
    if vulnerable:
        log.warning("⚠️  Version %s is **vulnerable** to CVE‑2024‑XXXXX.", ".".join(map(str, version)))
    else:
        log.info("✅ Version %s is NOT in the known vulnerable range.", ".".join(map(str, version)))
        # Still check migration setting for hygiene
        # (optional: you may want to leave it enabled if you're sure you're safe)
    # 3️⃣ Check migration setting
    try:
        migration_on = get_migration_enabled()
    except Exception as exc:
        log.error("Could not read migration setting: %s", exc)
        sys.exit(1)

    if migration_on:
        log.info("🔧 Repository migration is currently **ENABLED**.")
    else:
        log.info("🔒 Repository migration is already **DISABLED**.")

    # 4️⃣ Mitigation: if vulnerable AND migration enabled → disable it
    if vulnerable and migration_on:
        log.info("🛡️  Applying mitigation: disabling migration …")
        try:
            disable_migration()
        except Exception as exc:
            log.error("Mitigation failed: %s", exc)
            sys.exit(1)
        log.info("✅ Mitigation applied successfully.")
    elif vulnerable and not migration_on:
        log.info("✅ Instance is vulnerable but migration is already disabled – no action needed.")
    else:
        log.info("🎉 No action required.")


if __name__ == "__main__":
    main()

# Create .env (see Configuration section)
cp .env.example .env   # then edit with your values
python detect_and_mitigate_gitea.py

/**
 * detect-and-mitigate-gitea.ts
 *
 * Detects a vulnerable Gitea version and disables the repository migration
 * feature via the admin API if needed.
 *
 * Prerequisites:
 *   npm install axios dotenv
 *   npx ts-node detect-and-mitigate-gitea.ts   (or compile to JS)
 */

import * as dotenv from "dotenv";
import axios from "axios";
import { log, info, warn, error } from "./logger"; // simple logger wrapper (see below)

dotenv.config();

const GITEA_URL: string = (process.env.GITEA_URL ?? "").replace(/\/+$/, "");
const GITEA_TOKEN: string = process.env.GITEA_TOKEN ?? "";

if (!GITEA_URL || !GITEA_TOKEN) {
  console.error("❌ Error: GITEA_URL and GITEA_TOKEN must be set in .env");
  process.exit(1);
}

const api = axios.create({
  baseURL: GITEA_URL,
  headers: {
    Authorization: `token ${GITEA_TOKEN}`,
    Accept: "application/json",
  },
  timeout: 10_000,
});

/**
 * Simple logger – you can replace with winston/pino in production.
 */
function log(level: string, msg: string) {
  const timestamp = new Date().toISOString();
  console.log(`[${timestamp}] [${level.toUpperCase()}] ${msg}`);
}
const info = (msg: string) => log("info", msg);
const warn = (msg: string) => log("warn", msg);
const error = (msg: string) => log("error", msg);

/**
 * GET helper with error handling.
 */
async function get<T>(url: string): Promise<T> {
  try {
    const { data } = await api.get<T>(url);
    return data;
  } catch (err: any) {
    error(`GET ${url} failed: ${err.message}`);
    throw err;
  }
}

/**
 * PATCH helper with error handling.
 */
async function patch<T>(url: string, data: any): Promise<T> {
  try {
    const { data: resp } = await api.patch<T>(url, data);
    return resp;
  } catch (err: any) {
    error(`PATCH ${url} failed: ${err.message}`);
    throw err;
  }
}

/**
 * Parse version string "1.14.2" or "1.15.0-dev" => [1,14,2]
 */
function parseVersion(vstr: string): number[] {
  const clean = vstr.split("-")[0]; // strip pre-release
  return clean.split(".").map((p) => parseInt(p, 10));
}

/**
 * Check if version falls in known vulnerable ranges.
 */
function isVulnerable(ver: number[]): boolean {
  const [major, minor, patch] = ver;
  if (major !== 1) return false;
  if (minor === 13 && patch >= 0 && patch <= 6) return true;
  if (minor === 14 && patch >= 0 && patch <= 2) return true;
  if (minor === 15 && patch === 0) return true; // covers 1.15.0-dev treated as patch 0
  return false;
}

/**
}

/**
 * Fetch the migration setting.
 */
async function getMigrationEnabled(): Promise<boolean> {
  const settings = await get<any>("/api/v1/admin/settings");
  const migrate = settings?.repository?.migrate?.enable;
  if (typeof migrate === "boolean") return migrate;
  if (typeof migrate === "string") return migrate.toLowerCase() === "true";
  throw new Error("Unable to determine migration setting type");
}

/**
 * Disable migration feature.
 */
async function disableMigration(): Promise<void> {
  await patch<{ ok: boolean }>("/api/v1/admin/settings", {
    "repository.migrate.enable": false,
  });
  info("🔒 Migration feature disabled.");
}

/**
 * Main routine.
 */
async function main(): Promise<void> {
  info(`🔍 Checking Gitea instance at ${GITEA_URL}`);

  // 1️⃣ Version
  const versionInfo = await get<{ version: string }>("/api/v1/version");
  const versionStr = versionInfo.version;
  const version = parseVersion(versionStr);
  info(`📦 Detected Gitea version: ${versionStr}`);

  // 2️⃣ Vulnerability check
  const vulnerable = isVulnerable(version);
  if (vulnerable) {
    warn(`⚠️  Version ${versionStr} is **vulnerable** to CVE‑2024‑XXXXX.`);
  } else {
    info(`✅ Version ${versionStr} is NOT in the known vulnerable range.`);
  }

  // 3️⃣ Migration setting
  let migrationOn: boolean;
  try {
    migrationOn = await getMigrationEnabled();
  } catch (e: any) {
    error(`Failed to read migration setting: ${e.message}`);
    process.exit(1);
  }
  if (migrationOn) {
    info("🔧 Repository migration is currently **ENABLED**.");
  } else {
    info("🔒 Repository migration is already **DISABLED**.");
  }

  // 4️⃣ Mitigation
  if (vulnerable && migrationOn) {
    info("🛡️  Applying mitigation: disabling migration …");
    try {
      await disableMigration();
      info("✅ Mitigation applied successfully.");
    } catch (err: any) {
      error(`Mitigation failed: ${err.message}`);
      process.exit(1);
    }
  } else if (vulnerable && !migrationOn) {
    info("✅ Instance is vulnerable but migration is already disabled – no action needed.");
  } else {
    info("🎉 No action required.");
  }
}

// Run
main().catch((err) => {
  error(`Unhandled error: ${err}`);
  process.exit(1);
});

// logger.ts
export function log(level: string, msg: string) {
  const ts = new Date().toISOString();
  console.log(`[${ts}] [${level.toUpperCase()}] ${msg}`);
}
export const info = (msg: string) => log("info", msg);
export const warn = (msg: string) => log("warn", msg);
export const error = (msg: string) => log("error", msg);

# Create .env (see Configuration section)
cp .env.example .env   # then edit with your values
npx ts-node detect-and-mitigate-gitea.ts
# Or compile first:
# npx tsc detect-and-mitigate-gitea.ts && node detect-and-mitigate-gitea.js

# .env – keep this file secret!
# Base URL of your Gitea instance (no trailing slash)
GITEA_URL=https://git.example.com

# Personal access token with **admin** scope.
# Create it via: User Settings → Applications → Generate New Token
GITEA_TOKEN=ghp_******************************

def paginate_get(endpoint: str, params: dict = None) -> list:
    """Yield all items from a paginated Gitea endpoint."""
    if params is None:
        params = {}
    params.setdefault("limit", 100)
    page = 1
    all_items = []
    while True:
        params["page"] = page
        data = get_json(f"{GITEA_URL}/api/v1/{endpoint}", params)
        if not isinstance(data, list) or not data:
            break
        all_items.extend(data)
        # If we got fewer than limit, we're done
        if len(data) < params["limit"]:
            break
        page += 1
    return all_items

import { delay } from "await-delay"; // npm i await-delay

async function retry<T>(fn: () => Promise<T>, attempts: number = 3): Promise<T> {
  let lastErr: any;
  for (let i = 0; i < attempts; i++) {
    try {
      return await fn();
    } catch (err) {
      lastErr = err;
      if (i === attempts - 1) throw err;
      const backoff = Math.pow(2, i) * 300; // 300ms, 600ms, 1200ms
      warn(`Attempt ${i + 1} failed, retrying in ${backoff}ms…`);
      await delay(backoff);
    }
  }
  throw lastErr;
}

def request_with_rate_limit(method, url, **kwargs):
    while True:
        resp = requests.request(method, url, headers=HEADERS, **kwargs)
        if resp.status_code == 429:
            retry_after = int(resp.headers.get("Retry-After", "5"))
            warn(f"Rate limited – sleeping {retry_after}s")
            time.sleep(retry_after)
            continue
        resp.raise_for_status()
        return resp