gador/nixpkgs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

nixosTests.ec2-image: replace micro_httpd with IMDSv2-aware server

Browse Source 
micro_httpd returns 501 for PUT requests, so IMDSv2 token acquisition
was never actually tested. Replace it with a minimal Python IMDS server
that handles the full IMDSv2 flow: PUT for token, token validation on
GET requests, and file serving from a metadata directory.

This means the test now validates that:
- The fetcher correctly obtains an IMDSv2 token via PUT
- The token is passed on subsequent metadata GET requests
- Requests without a valid token are rejected (401)
This commit is contained in:
Philip Taron
2026-03-24 15:38:20 -07:00
parent 81dce7843c
commit c66b3b2d55
2 changed files with 119 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nixos/tests/common/imds-server.py
+95
View File
@@ -0,0 +1,95 @@
"""Minimal IMDSv2-compatible metadata server for NixOS EC2 tests.
Runs in inetd mode: reads one HTTP request from stdin, writes the
response to stdout. Drop-in replacement for micro_httpd in QEMU
guestfwd and socat contexts.
Usage: imds-server <metadata-directory>
The metadata directory should contain:
latest/api/token - Token value (returned on PUT)
1.0/meta-data/hostname - Instance hostname
1.0/meta-data/ami-manifest-path - AMI manifest path
1.0/meta-data/instance-id - Instance ID
1.0/meta-data/public-keys/0/openssh-key - SSH public key
1.0/user-data - User data
"""
import os
import sys
def read_request():
"""Read and parse one HTTP request from stdin (inetd mode)."""
request_line = sys.stdin.readline()
if not request_line:
sys.exit(0)
parts = request_line.strip().split()
method = parts[0] if parts else ""
path = parts[1] if len(parts) > 1 else "/"
headers = {}
while True:
line = sys.stdin.readline()
if not line or line.strip() == "":
break
if ":" in line:
key, _, value = line.partition(":")
headers[key.strip().lower()] = value.strip()
return method, path, headers
def respond(status, body):
"""Write an HTTP response to stdout."""
if isinstance(body, str):
body = body.encode()
header = (
f"HTTP/1.1 {status}\r\n"
f"Content-Type: text/plain\r\n"
f"Content-Length: {len(body)}\r\n"
f"Connection: close\r\n"
f"\r\n"
).encode()
sys.stdout.buffer.write(header + body)
sys.stdout.buffer.flush()
def main():
base_dir = sys.argv[1] if len(sys.argv) > 1 else "."
# Load expected token from file
token_path = os.path.join(base_dir, "latest", "api", "token")
if os.path.isfile(token_path):
with open(token_path) as f:
expected_token = f.read().strip()
else:
expected_token = "test-imdsv2-token"
method, path, headers = read_request()
rel_path = path.lstrip("/")
# PUT /latest/api/token — IMDSv2 token acquisition
if method == "PUT" and rel_path == "latest/api/token":
respond("200 OK", expected_token)
return
# All other requests require a valid token
request_token = headers.get("x-aws-ec2-metadata-token", "")
if request_token != expected_token:
respond("401 Unauthorized", "Invalid or missing IMDSv2 token\n")
return
# Serve file from the metadata directory
file_path = os.path.join(base_dir, rel_path)
if os.path.isfile(file_path):
with open(file_path, "rb") as f:
content = f.read()
respond("200 OK", content)
else:
respond("404 Not Found", f"Not found: {path}\n")
if __name__ == "__main__":
main()
nixos/tests/ec2-image.nix
+24 -12
View File
@@ -13,6 +13,11 @@ let
inherit (lib) mkAfter mkForce;
pkgs = config.node.pkgs;
# Minimal IMDSv2-compatible metadata server (inetd-mode, drop-in for micro_httpd)
imdsServer = pkgs.writers.writePython3Bin "imds-server" { } (
builtins.readFile ./common/imds-server.py
);
# Build an EC2 image configuration
imageCfg =
(import ../lib/eval-config.nix {
@@ -47,7 +52,7 @@ let
# Packages needed for IPv6 IMDS fallback test
environment.systemPackages = [
pkgs.socat
pkgs.micro-httpd
imdsServer
pkgs.iptables
];
@@ -94,11 +99,8 @@ in
# Instance Metadata Service (IMDSv2 with 1.0 metadata version)
# TODO: Use 'latest' metadata version instead of '1.0'
# - Consider https://github.com/aws/amazon-ec2-metadata-mock
# - Blocked on https://github.com/aws/amazon-ec2-metadata-mock/issues/234
# - Consider https://github.com/purpleclay/imds-mock
# - [Test matrix] also test providing the host key through IMDS
# - i.e. a test module argument to select between writing or reading the host key
# TODO: [Test matrix] also test providing the host key through IMDS
# - i.e. a test module argument to select between writing or reading the host key
def create_ec2_metadata_dir(temp_dir, client_pubkey):
"""Create fake EC2 metadata directory structure with mock data"""
metadata_dir = os.path.join(temp_dir.name, "ec2-metadata")
@@ -178,7 +180,7 @@ in
)
metadata_net = (
" -device virtio-net-pci,netdev=ec2meta"
+ f" -netdev 'user,id=ec2meta,net=169.0.0.0/8,guestfwd=tcp:169.254.169.254:80-cmd:${pkgs.micro-httpd}/bin/micro_httpd {metadata_dir}'"
+ f" -netdev 'user,id=ec2meta,net=169.0.0.0/8,guestfwd=tcp:169.254.169.254:80-cmd:${lib.getExe imdsServer} {metadata_dir}'"
)
start_command = (
@@ -227,7 +229,16 @@ in
machine_ip = "${config.nodes.machine.networking.primaryIPAddress}"
with subtest("EC2 metadata service connectivity"):
hostname_response = machine.succeed("curl --fail -s http://169.254.169.254/1.0/meta-data/hostname")
# Obtain an IMDSv2 token, then use it to fetch metadata
imds_token = machine.succeed(
"curl -sf -X PUT -H 'X-aws-ec2-metadata-token-ttl-seconds: 600'"
" http://169.254.169.254/latest/api/token"
).strip()
assert imds_token, "Failed to obtain IMDSv2 token"
hostname_response = machine.succeed(
f"curl -sf -H 'X-aws-ec2-metadata-token: {imds_token}'"
" http://169.254.169.254/1.0/meta-data/hostname"
)
assert "test-instance" in hostname_response, f"Expected 'test-instance', got: {hostname_response}"
with subtest("SSH host key extraction from console"):
@@ -329,16 +340,17 @@ in
" /tmp/ipv6-metadata/1.0/meta-data/public-keys/0/openssh-key"
)
# Serve metadata on the IPv6 IMDS address via socat + micro_httpd (inetd-style)
# Serve metadata on the IPv6 IMDS address via socat + imds-server (inetd-style)
machine.succeed(
"systemd-run --unit=ipv6-imds --"
" socat TCP6-LISTEN:80,bind=[fd00:ec2::254],fork,reuseaddr"
" SYSTEM:'${lib.getExe pkgs.micro-httpd} /tmp/ipv6-metadata'"
" SYSTEM:'${lib.getExe imdsServer} /tmp/ipv6-metadata'"
)
# Wait for IPv6 IMDS to become reachable
# Wait for IPv6 IMDS to become reachable (token endpoint doesn't require auth)
machine.wait_until_succeeds(
"curl -sf http://[fd00:ec2::254]/1.0/meta-data/hostname"
"curl -sf -X PUT -H 'X-aws-ec2-metadata-token-ttl-seconds: 600'"
" http://[fd00:ec2::254]/latest/api/token"
)
# Block IPv4 IMDS to force fallback to IPv6