1.5.7 pwntools 4.14.0 improvements

+ removed pwntools debug patch (pwntools==4.14.0)
+ fixed ssh port forwarding for alpine
+ fixed libc debug symbols (added new sysroot_debug args)
+ added Optional type hints

Author: gfelber

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "vagd"
-version = "1.5.5"
+version = "1.5.6"
 authors = [{ name = "0x6fe1be2" }]
 description = "VirtuAlization GDb integrations in pwntools"
 readme = "README.md"

requirements.txt

@@ -1,3 +1,3 @@
-pwntools
+pwntools==4.14.0
 docker
-typer
+typer

src/vagd/cli.py

@@ -2,7 +2,7 @@
 import os
 import stat
 import sys
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Any
 
 import typer
 from rich.console import Console
@@ -16,14 +16,14 @@
 from vagd.virts.vagd import Vagd
 
 DOGD_BOX = "Box.DOCKER_UBUNTU"
-DOGD = "vm = Dogd(BINARY, image={box}, {args})  # Docker"
+DOGD = "vm = Dogd(BINARY, image=BOX, {args})  # Docker"
 QEGD_BOX = "Box.QEMU_UBUNTU"
-QEGD = "vm = Qegd(BINARY, img={box}, {args})  # Qemu"
+QEGD = "vm = Qegd(BINARY, img=BOX, {args})  # Qemu"
 SHGD = "vm = Shgd(BINARY, user='user', host='localhost', port=22, {args})  # SSH"
 
 # deprecated
 VAGD_BOX = "Box.VAGRANT_JAMMY64"
-VAGD = "vm = Vagd(BINARY, {box}, {args})  # Vagrant"
+VAGD = "vm = Vagd(BINARY, BOX, {args})  # Vagrant"
 
 AD_ENV = """# ad envs
 IS_AD  = os.getenv('TARGET_IP') is not None           # running on ad
@@ -35,7 +35,9 @@
 err_console = Console(stderr=True)
 console = Console()
 
-quote = lambda x: f"'{x}'"
+
+def quote(x: str):
+  return f"'{x}'"
 
 
 def _version(value: bool) -> None:
@@ -65,12 +67,11 @@ def add_virt(
   dependency: str,
   template: str,
   args: Dict[str, str],
-  multi=False,
-  box="",
+  multi: bool = False,
 ):
   dependencies.append(dependency)
   args_str = ", ".join(f"{k}={v}" for k, v in args.items())
-  vm = template.format(box=box, args=args_str)
+  vm = template.format(args=args_str)
   vms.append(("# " if multi else "") + vm)
 
 
@@ -162,14 +163,14 @@ def template(
   if sum((dogd, qegd, vagd, shgd)) > 1:
     multi = True
 
-  env = {}
+  env: Dict[str, str] = dict()
 
   if libc:
     files.append(libc)
 
-  dependencies = []
-  vms = []
-  args = dict()
+  dependencies: list[str] = list()
+  vms: list[str] = list()
+  args: Dict[str, Any] = dict()
 
   if root:
     args["user"] = "'root'"
@@ -188,13 +189,17 @@ def template(
 
   args["ex"] = "True"
   args["fast"] = "True"
+  box = ""
 
   if dogd:
-    add_virt(dependencies, vms, "Dogd", DOGD, args, box=image)
+    box = image
+    add_virt(dependencies, vms, "Dogd", DOGD, args)
   if qegd:
-    add_virt(dependencies, vms, "Qegd", QEGD, args, multi, box=img)
+    box = img
+    add_virt(dependencies, vms, "Qegd", QEGD, args, multi)
   if vagd:
-    add_virt(dependencies, vms, "Vagd", VAGD, args, multi, box=vbox)
+    box = vbox
+    add_virt(dependencies, vms, "Vagd", VAGD, args, multi)
   if shgd:
     add_virt(dependencies, vms, "Shgd", SHGD, args, multi)
 
@@ -226,6 +231,7 @@ def template(
     ip=quote(ip),
     port=str(port),
     env=repr(env),
+    box=box,
     ad_env=AD_ENV if ad else "",
     vms=("\n" + " " * 4).join(vms),
     libc=quote(libc),
@@ -267,7 +273,7 @@ def _get_type() -> str:
   exit(1)
 
 
-def _exec(cmd: str, env: Dict = None):
+def _exec(cmd: str, env: Optional[Dict[str, str]] = None):
   if env is None:
     env = os.environ
   else:

src/vagd/res/aliases.txt

@@ -7,8 +7,8 @@ linfo = lambda x, *a: log.info(x, *a)
 lwarn = lambda x, *a: log.warn(x, *a)
 lerr  = lambda x, *a: log.error(x, *a)
 lprog = lambda x, *a: log.progress(x, *a)
-lhex  = lambda x, y="leak": linfo(f"0x{x:016x} <- {y}")
-phex  = lambda x, y="leak": print(f"0x{x:016x} <- {y}")
+lhex  = lambda x, y="leak": linfo(f"{x:#016x} <- {y}")
+phex  = lambda x, y="leak": print(f"{x:#016x} <- {y}")
 
 # type manipulation
 byt   = lambda x: x if isinstance(x, (bytes, bytearray)) else f"{x}".encode()
@@ -41,7 +41,7 @@ ss  = (
         if len(x) < s
         else se(x, *a, **kw)
           if len(x) == s
-          else lerr("ss to big: 0x%x > 0x%x", len(x), s)
+          else lerr(f"ss to big: {len(x):#x} > {s:#x}")
       )
 sla = lambda x, y, t=None, *a, **kw: gt(t).sendlineafter(
         byt(x), byt(y), *a, **kw
@@ -52,7 +52,7 @@ sas = (
         if len(y) < s
         else sa(x, y, *a, **kw)
           if len(y) == s
-          else lerr("sas to big: 0x%x > 0x%x", len(y), s)
+          else lerr(f"ss to big: {len(x):#x} > {s:#x}")
       )
 ra  = lambda t=None, *a, **kw: gt(t).recvall(*a, **kw)
 rl  = lambda t=None, *a, **kw: gt(t).recvline(*a, **kw)

src/vagd/res/template.txt

@@ -11,6 +11,7 @@ PORT   = {port:<44s} # remote PORT
 BINARY = {binary:<44s} # PATH to local binary
 ARGS   = []                                           # ARGS supplied to binary
 ENV    = {env:<44s} # ENV supplied to binary
+BOX    = {box:<44s} # Docker box image
 {ad_env}
 # GDB SCRIPT, executed at start of GDB session (e.g. set breakpoints here)
 GDB    = f"""
@@ -25,7 +26,7 @@ context.aslr = {aslr:<5s}                                  # ASLR enabled (only
 
 # setup vagd vm
 vm = None
-def setup() -> object | None:
+def setup():
   global vm
   if args.REMOTE or {is_local}{is_ad}:
     return None
@@ -45,7 +46,7 @@ def setup() -> object | None:
 
 
 # get target (pwnlib.tubes.tube)
-def get_target(**kw) -> tubes.tube:
+def get_target(**kw):
   if args.REMOTE{is_ad}:
     # context.log_level = 'debug'
     return remote(IP, PORT)

src/vagd/templates.py

@@ -63,6 +63,7 @@
 RUN adduser -h /home/{user} -s /bin/ash -g sudo -D {user}
 RUN echo "{user} ALL=(ALL) NOPASSWD:ALL" > /etc/sudoers.d/{user} && chmod 0440 /etc/sudoers.d/{user}
 RUN echo "{user}:{user}" | chpasswd
+RUN sed "s/AllowTcpForwarding no/AllowTcpForwarding yes/" -i /etc/ssh/sshd_config
 
 USER {user}
 

src/vagd/virts/dogd.py

@@ -1,5 +1,5 @@
 import os
-from typing import Dict, List
+from typing import Dict, List, Optional
 
 import docker
 
@@ -67,7 +67,6 @@ class Dogd(Shgd):
   _dockerdir: str
   _dockerfile: str
   _isalpine: bool
-  _gdbsrvport: int
   _rm: bool
   _ex: bool
   _forward: Dict[str, int]
@@ -88,8 +87,8 @@ def __init__(
     binary: str,
     image: str = DEFAULT_IMAGE,
     user: str = DEFAULT_USER,
-    forward: Dict[str, int] = None,
-    packages: List[str] = None,
+    forward: Optional[Dict[str, int]] = None,
+    packages: Optional[List[str]] = None,
     symbols=True,
     rm=True,
     ex: bool = False,
@@ -116,11 +115,8 @@ def __init__(
       # trigger package detection in Pwngdb
       packages = list()
 
-    self._gdbsrvport = -1
     self._dockerdir = Dogd.DOCKERHOME + f"{self._image}/"
-    if not (
-      os.path.exists(Dogd.DOCKERHOME) and os.path.exists(self._dockerdir)
-    ):
+    if not (os.path.exists(Dogd.DOCKERHOME) and os.path.exists(self._dockerdir)):
       os.makedirs(self._dockerdir)
     self._dockerfile = self._dockerdir + "Dockerfile"
     self._user = user
@@ -143,7 +139,6 @@ def __init__(
       ex=ex,
       fast=fast,
       symbols=False,
-      gdbsrvport=self._gdbsrvport,
       **kwargs,
     )
 
@@ -154,11 +149,7 @@ def _create_dockerfile(self):
 
     if not os.path.exists(self._dockerdir + "keyfile.pub"):
       os.link(Pwngd.PUBKEYFILE, self._dockerdir + "keyfile.pub")
-    template = (
-      templates.DOCKER_ALPINE_TEMPLATE
-      if self._isalpine
-      else templates.DOCKER_TEMPLATE
-    )
+    template = templates.DOCKER_ALPINE_TEMPLATE if self._isalpine else templates.DOCKER_TEMPLATE
 
     with open(self._dockerfile, "w") as dockerfile:
       dockerfile.write(
@@ -175,9 +166,6 @@ def _create_docker_instance(self):
     helper.info("starting docker instance")
     self._port = helper.first_free_port(Dogd.DEFAULT_PORT)
     self._forward.update({"22/tcp": self._port})
-    if self._isalpine:
-      self._gdbsrvport = helper.first_free_port(Pwngd.STATIC_GDBSRV_PORT)
-      self._forward.update({f"{self._gdbsrvport}/tcp": self._gdbsrvport})
 
     dir = os.path.dirname(os.path.realpath(__file__))
     with open(dir[: dir.rfind("/")] + "/res/seccomp.json", "r") as seccomp_file:
@@ -194,9 +182,7 @@ def _create_docker_instance(self):
     self._id = container.id
     helper.info(f"started docker instance {container.short_id}")
     with open(Dogd.LOCKFILE, "w") as lockfile:
-      lockfile.write(
-        f"{container.id}:{str(self._port)}:{str(self._gdbsrvport)}"
-      )
+      lockfile.write(f"{container.id}:{str(self._port)}")
 
   def _build_image(self):
     build_progress = helper.progress("building docker image")
@@ -215,9 +201,7 @@ def _build_image(self):
         tag += "_"
       tag += "symbols"
 
-    bimage = self._client.images.build(
-      path=os.path.dirname(self._dockerfile), tag=f"vagd/{tag}"
-    )[0]
+    bimage = self._client.images.build(path=os.path.dirname(self._dockerfile), tag=f"vagd/{tag}")[0]
 
     build_progress.success("done")
 
@@ -239,21 +223,15 @@ def _vm_create(self):
   def _vm_setup(self) -> None:
     self._client = docker.from_env()
     if not os.path.exists(Dogd.LOCKFILE):
-      helper.info(
-        f"No Lockfile {Dogd.LOCKFILE} found, creating new Docker Instance"
-      )
+      helper.info(f"No Lockfile {Dogd.LOCKFILE} found, creating new Docker Instance")
       self._vm_create()
     else:
       with open(Dogd.LOCKFILE, "r") as lockfile:
         data = lockfile.readline().split(":")
         self._id = data[0]
         self._port = int(data[1])
-        if self._isalpine:
-          self._gdbsrvport = int(data[2])
       if not self._client.containers.list(filters={"id": self._id}):
-        helper.info(
-          f"Lockfile {Dogd.LOCKFILE} found, container not running, creating new one"
-        )
+        helper.info(f"Lockfile {Dogd.LOCKFILE} found, container not running, creating new one")
         self._vm_create()
       else:
         helper.info(

src/vagd/virts/logd.py

@@ -1,4 +1,4 @@
-from typing import Iterable
+from typing import Iterable, Optional
 
 import pwnlib.args
 import pwnlib.tubes
@@ -64,7 +64,7 @@ def _install_packages(self, packages: Iterable):
     """
     helper.error("NOT IMPLEMENTED")
 
-  def put(self, file: str, remote: str = None):
+  def put(self, file: str, remote: Optional[str] = None):
     """
     NOT IMPLEMENTED
     """
@@ -78,9 +78,7 @@ def debug(self, **kwargs) -> pwnlib.tubes.process.process:
     """
     return self.pwn_debug(**kwargs)
 
-  def pwn_debug(
-    self, argv: list[str] = None, **kwargs
-  ) -> pwnlib.tubes.process.process:
+  def pwn_debug(self, argv: Optional[list[str]] = None, **kwargs) -> pwnlib.tubes.process.process:
     """
     run binary with gdb locally
     :param argv: comandline arguments for binary
@@ -89,9 +87,7 @@ def pwn_debug(
     """
     return pwnlib.gdb.debug([self._binary] + argv, **kwargs)
 
-  def process(
-    self, argv: list[str] = None, **kwargs
-  ) -> pwnlib.tubes.process.process:
+  def process(self, argv: Optional[list[str]] = None, **kwargs) -> pwnlib.tubes.process.process:
     """
     run binary locally
     :param argv: comandline arguments for binary
@@ -102,9 +98,9 @@ def process(
 
   def start(
     self,
-    argv: list[str] = None,
+    argv: Optional[list[str]] = None,
     gdbscript: str = "",
-    api: bool = None,
+    api: bool = False,
     **kwargs,
   ) -> pwnlib.tubes.process.process:
     """