#!/usr/bin/env python3
"""Real Tor SOCKS request from a disposable peer network namespace.

Run: sudo -n unshare --net python3 apps/lnd-tor/test_namespace.py
No host routes, service commands, wallet mounts or external network access.
"""
import ctypes
import os
from pathlib import Path
import socket
import subprocess
import tempfile
import time


def ip(*args):
    subprocess.run(["ip", *args], check=True)


def main():
    # Refuse to operate in the host network. Also supports the locked PID1
    # fixture wrapper used by the combined backend verification run.
    assert os.getpid() == 1 or os.stat("/proc/self/ns/net").st_ino != os.stat("/proc/1/ns/net").st_ino, "requires disposable network namespace"
    ip("link", "set", "lo", "up")
    ready, child_ready = socket.socketpair()
    pid = os.fork()
    if pid == 0:
        ready.close()
        try:
            assert ctypes.CDLL(None, use_errno=True).unshare(0x40000000) == 0
            child_ready.sendall(b"N")
            assert child_ready.recv(1) == b"G"
            ip("link", "set", "lo", "up")
            ip("addr", "add", "10.89.0.2/24", "dev", "requester")
            ip("addr", "add", "198.51.100.2/24", "dev", "requester")
            ip("link", "set", "requester", "up")
            # Host loopback cannot serve this request namespace.
            try:
                socket.create_connection(("127.0.0.1", 9050), 1)
                raise AssertionError("unexpected requester-loopback SOCKS")
            except ConnectionRefusedError:
                pass
            with socket.create_connection(("10.89.0.1", 9050), 3) as client:
                client.sendall(b"\x05\x01\x00")
                assert client.recv(2) == b"\x05\x00", "SOCKS5 negotiation failed"
            # A non-private source on the same disposable link is denied.
            with socket.create_connection(("198.51.100.1", 9050), 3) as client:
                try:
                    client.sendall(b"\x05\x01\x00")
                    assert client.recv(2) != b"\x05\x00", "public source accepted"
                except ConnectionResetError:
                    pass
            child_ready.sendall(b"P")
            os._exit(0)
        except BaseException:
            import traceback
            traceback.print_exc()
            os._exit(1)
    child_ready.close()
    tor = None
    try:
        assert ready.recv(1) == b"N"
        ip("link", "add", "transport", "type", "veth", "peer", "name", "requester")
        ip("link", "set", "requester", "netns", str(pid))
        ip("addr", "add", "10.89.0.1/24", "dev", "transport")
        ip("addr", "add", "198.51.100.1/24", "dev", "transport")
        ip("link", "set", "transport", "up")
        with tempfile.TemporaryDirectory(prefix="lnd-tor-fixture-") as tmp:
            with open(Path(tmp) / "tor.log", "w+") as log:
                tor = subprocess.Popen(["tor", "-f", str(Path(__file__).with_name("torrc")),
                    "--DataDirectory", tmp], stdout=log, stderr=log)
                for _ in range(100):
                    try:
                        with socket.create_connection(("127.0.0.1", 9050), .1):
                            break
                    except OSError:
                        assert tor.poll() is None, "disposable Tor exited"
                        time.sleep(.05)
                else:
                    raise AssertionError("Tor SOCKS listener never became ready")
                ready.sendall(b"G")
                ready.settimeout(10)
                assert ready.recv(1) == b"P"
                assert os.waitpid(pid, 0)[1] == 0
                print("PASS: real Tor SOCKS5 reachable across disposable request namespace; requester loopback refused; non-private source denied", flush=True)
    finally:
        if tor is not None:
            tor.terminate()
            tor.wait(timeout=10)
        ready.close()


if __name__ == "__main__":
    main()