mirror of
https://github.com/mrvasil/ctf-ad-mirror-tool-rust.git
synced 2026-10-23 14:31:02 +00:00
79 lines
2.2 KiB
Python
79 lines
2.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Disposable loopback-only UI fixture. Never reads legacy configuration.
|
|
|
|
Usage: python3 scripts/ui_fixture.py target/debug/mirror-tool
|
|
Public test-only login: local-ui-fixture-only. Ctrl-C removes fixture state.
|
|
"""
|
|
import asyncio
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import sys
|
|
import tempfile
|
|
|
|
|
|
async def echo(reader, writer):
|
|
try:
|
|
while data := await reader.read(65536):
|
|
writer.write(data)
|
|
await writer.drain()
|
|
except (OSError, asyncio.CancelledError):
|
|
pass
|
|
finally:
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
|
|
|
|
async def main():
|
|
executable = str(Path(sys.argv[1]).resolve())
|
|
with tempfile.TemporaryDirectory(prefix="mirror-ui-") as folder:
|
|
root = Path(folder)
|
|
config = root / "mirror.toml"
|
|
config.write_text('''[web]
|
|
listen = "127.0.0.1:16700"
|
|
state_file = "state.json"
|
|
[network]
|
|
drain_timeout_ms = 500
|
|
[[teams]]
|
|
id = "a"
|
|
name = "Loopback A"
|
|
address = "127.0.0.1"
|
|
[[teams]]
|
|
id = "b"
|
|
name = "Loopback B"
|
|
address = "::1"
|
|
[[services]]
|
|
id = "echo"
|
|
name = "Echo"
|
|
listen = "127.0.0.1:18081"
|
|
upstream_port = 18082
|
|
primary = "a"
|
|
replicas = ["b"]
|
|
enabled = false
|
|
auto_switch = false
|
|
replica_policy = "isolate"
|
|
''')
|
|
servers = [await asyncio.start_server(echo, address, 18082) for address in ("127.0.0.1", "::1")]
|
|
env = os.environ.copy()
|
|
env["MIRROR_PASSWORD"] = "local-ui-fixture-only"
|
|
process = await asyncio.create_subprocess_exec(executable, "--config", str(config), env=env)
|
|
stop = asyncio.Event()
|
|
for sig in (signal.SIGINT, signal.SIGTERM):
|
|
asyncio.get_running_loop().add_signal_handler(sig, stop.set)
|
|
print("UI fixture: http://127.0.0.1:16700 (loopback only)", flush=True)
|
|
waiter = asyncio.create_task(process.wait())
|
|
stopper = asyncio.create_task(stop.wait())
|
|
await asyncio.wait([waiter, stopper], return_when=asyncio.FIRST_COMPLETED)
|
|
if process.returncode is None:
|
|
process.terminate()
|
|
await process.wait()
|
|
stopper.cancel()
|
|
await asyncio.gather(stopper, return_exceptions=True)
|
|
for server in servers:
|
|
server.close()
|
|
await server.wait_closed()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|