mirror of
https://github.com/mrvasil/ctf-ad-mirror-tool-rust.git
synced 2026-10-24 22:49:25 +00:00
133 lines
6.2 KiB
Python
133 lines
6.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Test the shipped archive in isolation; never print generated credentials."""
|
|
import http.cookiejar
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import urllib.request
|
|
|
|
|
|
def free_port():
|
|
with socket.socket() as sock:
|
|
sock.bind(("127.0.0.1", 0))
|
|
return sock.getsockname()[1]
|
|
|
|
|
|
with tempfile.TemporaryDirectory(prefix="mirror package ") as temporary:
|
|
root = Path(temporary)
|
|
with tarfile.open(sys.argv[1]) as archive:
|
|
archive.extractall(root, filter="data")
|
|
package = root / "mirror-tool-linux-x64"
|
|
env = os.environ.copy()
|
|
env.pop("MIRROR_PASSWORD", None)
|
|
subprocess.run([str(package / "run.sh"), "check"], cwd=root, env=env, check=True)
|
|
config = package / "mirror.toml"
|
|
password = "mirror-tool-change-me"
|
|
assert f'password = "{password}"' in config.read_text()
|
|
assert 'listen = "0.0.0.0:6700"' in config.read_text()
|
|
assert stat.S_IMODE(config.stat().st_mode) == 0o600
|
|
assert not (package / "data/panel-password").exists()
|
|
# Explicit config must win over a legacy password file during upgrades.
|
|
legacy_password = "legacy-synthetic-passphrase"
|
|
(package / "data/panel-password").write_text(legacy_password)
|
|
web_port = free_port()
|
|
config.write_text(config.read_text().replace("0.0.0.0:6700", f"0.0.0.0:{web_port}"))
|
|
original_config = config.read_bytes()
|
|
base = f"http://127.0.0.1:{web_port}"
|
|
upstream = socket.socket()
|
|
upstream.bind(("127.0.0.1", 0))
|
|
upstream.listen()
|
|
upstream.settimeout(15)
|
|
received = bytearray()
|
|
|
|
def echo():
|
|
connection, _ = upstream.accept()
|
|
with connection:
|
|
connection.settimeout(10)
|
|
while data := connection.recv(65536):
|
|
received.extend(data)
|
|
connection.sendall(data)
|
|
|
|
thread = threading.Thread(target=echo, daemon=True)
|
|
thread.start()
|
|
proxy_port = free_port()
|
|
for phase in range(3):
|
|
restart = phase > 0
|
|
with (root / "server.log").open("wb") as log:
|
|
process = subprocess.Popen([str(package / "run.sh")], cwd=root, env=env, stdout=log, stderr=log)
|
|
try:
|
|
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()))
|
|
for _ in range(100):
|
|
if process.poll() is not None:
|
|
raise RuntimeError("packaged process exited during startup")
|
|
try:
|
|
with opener.open(base + "/", timeout=1) as response:
|
|
assert b"html" in response.read().lower()
|
|
break
|
|
except OSError:
|
|
time.sleep(0.1)
|
|
else:
|
|
raise RuntimeError("packaged server did not become ready")
|
|
assert any(line.split()[1] == f"00000000:{web_port:04X}" and line.split()[3] == "0A" for line in Path("/proc/net/tcp").read_text().splitlines()[1:])
|
|
|
|
def request(path, value=None, method=None, csrf=None):
|
|
headers = {"Content-Type": "application/json"}
|
|
if csrf:
|
|
headers["x-csrf-token"] = csrf
|
|
req = urllib.request.Request(base + path, data=None if value is None else json.dumps(value).encode(), headers=headers, method=method)
|
|
with opener.open(req, timeout=3) as response:
|
|
return json.load(response)
|
|
|
|
csrf = request("/api/login", {"password": password})["csrf"]
|
|
if not restart:
|
|
snapshot = request("/api/snapshot")
|
|
assert snapshot["teams"] == [] and snapshot["services"] == []
|
|
assert snapshot["flag_protection"] is None
|
|
request("/api/teams/local", {"id": "local", "name": "Local", "address": "127.0.0.1"}, "PUT", csrf)
|
|
service = {"id": "echo", "name": "Echo", "listen": f"127.0.0.1:{proxy_port}", "upstream_port": upstream.getsockname()[1], "primary": "local", "replicas": [], "enabled": True, "auto_switch": False}
|
|
request("/api/services/echo", service, "PUT", csrf)
|
|
payload = bytes(range(256)) * 1024
|
|
with socket.create_connection(("127.0.0.1", proxy_port), timeout=5) as connection:
|
|
connection.sendall(payload)
|
|
connection.shutdown(socket.SHUT_WR)
|
|
reply = bytearray()
|
|
while part := connection.recv(65536):
|
|
reply.extend(part)
|
|
thread.join(5)
|
|
assert not thread.is_alive()
|
|
assert reply == received == payload
|
|
snapshot = request("/api/snapshot")
|
|
assert password not in json.dumps(snapshot)
|
|
assert snapshot["services"][0]["flag_protection"] is False
|
|
assert snapshot["services"][0]["enabled"] is True
|
|
assert (package / "data/panel-password").read_text() == legacy_password
|
|
assert config.read_bytes() == original_config
|
|
finally:
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=20)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait()
|
|
raise
|
|
assert process.returncode == 0
|
|
if not restart:
|
|
password = "updated-config-passphrase"
|
|
config.write_text(config.read_text().replace("mirror-tool-change-me", password))
|
|
original_config = config.read_bytes()
|
|
elif phase == 1:
|
|
# Old v0.1.0 config without web.password still uses the saved password.
|
|
config.write_text(config.read_text().replace(f'password = "{password}"\n', ""))
|
|
password = legacy_password
|
|
original_config = config.read_bytes()
|
|
upstream.close()
|
|
print("Package OK: wildcard IPv4 listener, config password/change/restart, legacy-file precedence, empty topology, API persistence, exact 256KiB TCP half-close, flag substitution off.")
|