✨ Arbitrum Nitro node support (#2)

kubernetes/ctf-server.yaml

@@ -91,7 +91,7 @@ spec:
         runAsNonRoot: true
       containers:
       - name: orchestrator
-        image: us-docker.pkg.dev/idyllic-adviser-409615/openzeppelin/ctf-2023-server:latest
+        image: us-docker.pkg.dev/idyllic-adviser-409615/openzeppelin/ctf-2024-server:latest
         command: ["uvicorn", "--host", "0.0.0.0", "--port", "7283", "ctf_server:orchestrator"]
         env:
         - name: BACKEND
@@ -142,7 +142,7 @@ spec:
         runAsNonRoot: true
       containers:
       - name: anvil-proxy
-        image: us-docker.pkg.dev/idyllic-adviser-409615/openzeppelin/ctf-2023-server:latest
+        image: us-docker.pkg.dev/idyllic-adviser-409615/openzeppelin/ctf-2024-server:latest
         command: ["uvicorn", "--host", "0.0.0.0", "--port", "8545", "--workers", "16", "ctf_server:anvil_proxy"]
         env:
         - name: DATABASE

paradigmctf.py/Dockerfile

@@ -30,6 +30,20 @@ RUN true && \
     huffup && \
     true
 
+# Install Stylus
+
+env RUSTUP_HOME=/opt/rust/rustup CARGO_HOME=/opt/rust/cargo
+env PATH=${PATH}:/opt/rust/cargo/bin
+
+RUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y
+RUN rustup toolchain install nightly
+RUN rustup component add rust-src --toolchain nightly
+RUN rustup target add wasm32-unknown-unknown
+RUN RUSTFLAGS="-C link-args=-rdynamic" cargo install --force cargo-stylus
+RUN cargo install wasm-opt --locked
+
+RUN chmod -R 777 /opt/rust
+
 # (Optimization) Install requirements
 COPY requirements.txt /tmp/requirements.txt
 

paradigmctf.py/ctf_launchers/launcher.py

@@ -8,7 +8,7 @@
 
 import requests
 from ctf_launchers.team_provider import TeamProvider
-from ctf_launchers.utils import deploy, deploy_cairo, http_url_to_ws
+from ctf_launchers.utils import deploy, deploy_cairo, deploy_nitro, http_url_to_ws
 from ctf_server.types import (
     CreateInstanceRequest,
     DaemonInstanceArgs,
@@ -124,11 +124,14 @@ def launch_instance(self) -> int:
 
         if self.type == "starknet":
             web3 = get_privileged_web3(user_data, "main")
-            
+
             credentials = self.get_credentials(web3.provider.endpoint_uri)
 
             challenge_addr = self.deploy_cairo(user_data, credentials)
             priv_key = credentials[1][1]
+        elif self.type == "nitro":
+            challenge_addr = self.deploy_nitro(user_data, self.mnemonic)
+            priv_key = get_player_account(self.mnemonic).key.hex()
         else:
             challenge_addr = self.deploy(user_data, self.mnemonic)
             priv_key = get_player_account(self.mnemonic).key.hex()
@@ -141,7 +144,8 @@ def launch_instance(self) -> int:
 
         print()
         print(f"your private blockchain has been set up")
-        print(f"it will automatically terminate in {round(TIMEOUT/60)} minutes")
+        print(
+            f"it will automatically terminate in {round(TIMEOUT/60)} minutes")
         print(f"---")
         print(f"rpc endpoints:")
         for id in user_data["anvil_instances"]:
@@ -171,12 +175,19 @@ def deploy(self, user_data: UserData, mnemonic: str) -> str:
             web3, self.project_location, mnemonic, env=self.get_deployment_args(
                 user_data)
         )
-        
+
     def deploy_cairo(self, user_data: UserData, credentials: list) -> str:
         web3 = get_privileged_web3(user_data, "main")
 
         return deploy_cairo(web3, self.project_location, credentials, env=self.get_deployment_args(user_data))
-
+
+    def deploy_nitro(self, user_data: UserData, mnemonic: str) -> str:
+        web3 = get_privileged_web3(user_data, "main")
+
+        return deploy_nitro(
+            web3, self.project_location, mnemonic, env=self.get_deployment_args(
+                user_data)
+        )
 
     def get_deployment_args(self, user_data: UserData) -> Dict[str, str]:
         return {}

paradigmctf.py/ctf_launchers/nitro_pwn_launcher.py

@@ -0,0 +1,57 @@
+import os
+
+from eth_abi import abi
+import requests
+from ctf_launchers.launcher import Action, Launcher, ORCHESTRATOR_HOST, CHALLENGE
+from ctf_launchers.team_provider import TeamProvider, get_team_provider
+from ctf_server.types import UserData, get_privileged_web3
+from web3 import Web3
+
+FLAG = os.getenv("FLAG", "PCTF{flag}")
+
+
+class NitroPwnChallengeLauncher(Launcher):
+    def __init__(
+        self,
+        project_location: str = "challenge/project",
+        provider: TeamProvider = get_team_provider(),
+    ):
+        super().__init__(
+            'nitro',
+            project_location,
+            provider,
+            [
+                Action(name="get flag", handler=self.get_flag),
+            ],
+        )
+
+    def get_flag(self) -> int:
+        instance_body = requests.get(f"{ORCHESTRATOR_HOST}/instances/{self.get_instance_id()}").json()
+        if not instance_body['ok']:
+            print(instance_body['message'])
+            return 1
+
+        user_data = instance_body['data']
+
+        if not self.is_solved(
+            user_data, user_data['metadata']["challenge_address"]
+        ):
+            print("are you sure you solved it?")
+            return 1
+
+        print(FLAG)
+        return 0
+
+    def is_solved(self, user_data: UserData, addr: str) -> bool:
+        web3 = get_privileged_web3(user_data, "main")
+
+        (result,) = abi.decode(
+            ["bool"],
+            web3.eth.call(
+                {
+                    "to": addr,
+                    "data": web3.keccak(text="isSolved()")[:4],
+                }
+            ),
+        )
+        return result

paradigmctf.py/ctf_launchers/utils.py

@@ -1,16 +1,14 @@
 import json
 import os
-import shutil
+import re
 import subprocess
 from typing import Dict
 
-from eth_account.account import LocalAccount
 from web3 import Web3
 
 from foundry.anvil import anvil_autoImpersonateAccount, anvil_setCode
 
 
-
 def deploy(
     web3: Web3,
     project_location: str,
@@ -109,6 +107,143 @@ def deploy_cairo(
     return output[:65]
 
 
+def deploy_no_impersonate(
+    web3: Web3,
+    project_location: str,
+    mnemonic: str,
+    token: str,
+    deploy_script: str = "script/Deploy.s.sol:Deploy",
+    env: Dict = {}
+) -> str:
+    proc = subprocess.Popen(
+        args=[
+            "/opt/foundry/bin/forge",
+            "create",
+            "src/Challenge.sol:Challenge",
+            "--constructor-args",
+            token,
+            "--rpc-url",
+            web3.provider.endpoint_uri,
+            "--private-key",
+            "0xb6b15c8cb491557369f3c7d2c287b053eb229daa9c22138887752191c9520659"
+        ],
+        env={
+            "PATH": "/opt/huff/bin:/opt/foundry/bin:/usr/bin:" + os.getenv("PATH", "/fake"),
+        }
+        | env,
+        cwd=project_location,
+        text=True,
+        encoding="utf8",
+        stdin=subprocess.DEVNULL,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    stdout, stderr = proc.communicate()
+
+    if proc.returncode != 0:
+        print(stdout)
+        print(stderr)
+        raise Exception("forge failed to run")
+
+    address = stdout.split('Deployed to: ')[
+        1].replace("\\n", "")[:42]
+
+    cast_initialize(web3, project_location, token, address)
+
+    return address
+
+
+def cast_initialize(
+    web3: Web3,
+    project_location: str,
+    token: str,
+    entrypoint: str
+) -> str:
+    proc = subprocess.Popen(
+        args=[
+            "/opt/foundry/bin/cast",
+            "send",
+            token,
+            "0xc4d66de8000000000000000000000000"  + entrypoint[2:],
+            "--rpc-url",
+            web3.provider.endpoint_uri,
+            "--private-key",
+            "0xb6b15c8cb491557369f3c7d2c287b053eb229daa9c22138887752191c9520659"
+        ],
+        cwd=project_location,
+        text=True,
+        encoding="utf8",
+        stdin=subprocess.DEVNULL,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+
+    stdout, stderr = proc.communicate()
+
+    if proc.returncode != 0:
+        print(stdout)
+        print(stderr)
+        raise Exception("cast failed to run")
+
+
+def deploy_nitro(
+    web3: Web3,
+    project_location: str,
+    mnemonic: list,
+    env: Dict = {},
+) -> str:
+    rfd, wfd = os.pipe2(os.O_NONBLOCK)
+
+    proc = subprocess.Popen(
+        args=[
+            "/opt/rust/cargo/bin/cargo",
+            "stylus",
+            "deploy",
+            "-e",
+            web3.provider.endpoint_uri,
+            "--private-key",
+            "0xb6b15c8cb491557369f3c7d2c287b053eb229daa9c22138887752191c9520659"
+        ],
+        pass_fds=[wfd],
+        cwd=project_location,
+        text=True,
+        encoding="utf8",
+        stdin=subprocess.DEVNULL,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    stdout, stderr = proc.communicate()
+
+    if proc.returncode != 0:
+        print(stdout)
+        print(stderr)
+        raise Exception("script failed to run")
+
+    address = stdout.split('Activating program at address ')[
+        1].replace("\\n", "")
+
+    ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
+
+    token = ansi_escape.sub('', address)[:42]
+
+    env = {
+        "PATH": "/opt/huff/bin:/opt/foundry/bin:/usr/bin:" + os.getenv("PATH", "/fake"),
+        "MNEMONIC": mnemonic,
+        "OUTPUT_FILE": f"/proc/self/fd/{wfd}",
+        "TOKEN": token
+    }
+
+    output = deploy_no_impersonate(
+        web3,
+        project_location,
+        "",
+        token,
+        env=env,
+    )
+
+    return output
+
+
 def anvil_setCodeFromFile(
     web3: Web3,
     addr: str,
@@ -123,10 +258,11 @@ def anvil_setCodeFromFile(
 
     anvil_setCode(web3, addr, bytecode)
 
+
 def http_url_to_ws(url: str) -> str:
     if url.startswith("http://"):
-        return "ws://" + url[len("http://") :]
+        return "ws://" + url[len("http://"):]
     elif url.startswith("https://"):
-        return "wss://" + url[len("https://") :]
+        return "wss://" + url[len("https://"):]
 
     return url

paradigmctf.py/ctf_server/backends/backend.py

@@ -107,4 +107,31 @@ def _prepare_node_starknet(self, args: LaunchAnvilInstanceArgs, web3: Web3):
                 starknet_getVersion(web3)
                 break
             except:
-                time.sleep(0.1)
+                time.sleep(0.1)
+
+    def _prepare_node_nitro(self, args: LaunchAnvilInstanceArgs, web3: Web3):
+        while not web3.is_connected():
+            time.sleep(0.1)
+            continue
+
+        pk = "0xb6b15c8cb491557369f3c7d2c287b053eb229daa9c22138887752191c9520659"
+        acc = web3.eth.account.from_key(pk)
+
+        for i in range(args.get("accounts", DEFAULT_ACCOUNTS)):
+            transaction = {
+                'from': acc.address,
+                'to': self.__derive_account(
+                    args.get("derivation_path", DEFAULT_DERIVATION_PATH),
+                    args.get("mnemonic", DEFAULT_MNEMONIC),
+                    i,
+                ).address,
+                'value': 250 * 10 ** 18,
+                'nonce': web3.eth.get_transaction_count(acc.address),
+                'gas': 1000000,
+                'chainId': web3.eth.chain_id,
+                'gasPrice': web3.eth.gas_price
+            }
+
+            signed = web3.eth.account.sign_transaction(transaction, pk)
+
+            web3.eth.send_raw_transaction(signed.rawTransaction)