✨ Arbitrum Nitro node support

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)