Developer Guide

Developer Guide -- Internal API Reference

Internal API reference for developers who want to understand, modify, or extend the framework. This page documents every public function's signature, parameters, return types, exceptions, and calling conventions across all 21 source modules.

Function Categories

The framework is organized into these function groups:

1. Validation (validation.py) -- 9 whitelist input validators at trust boundary
2. Session Management (session.py) -- register, read, find, lock, cleanup, migrate
3. Command Builders (commands.py) -- construct socat argument lists
4. Process Management (process.py) -- launch (setsid), 9-step stop, port checks
5. Watchdog (watchdog.py) -- monitor-first auto-restart
6. Certificates (certs.py) -- TLS self-signed cert generation
7. Logging (logging_setup.py) -- structured dual-output logging
8. Configuration (config.py) -- constants, frozen dataclasses, protocol maps
9. CLI (cli.py) -- argparse with 10 subcommands
10. Menu (menu.py) -- interactive TUI with guided input
11. Mode Handlers (modes/*.py) -- one per operational mode

---

Validation Functions

All validation functions raise ValidationError (subclass of ValueError) on invalid input. They never silently correct or sanitize. Called by mode handlers and menu prompts before any command construction.

validate_port()

validate_port(port: str | int) -> int

Validate a port number is numeric and within valid range (1-65535).

• port: Port number as string or integer. Returns: Validated port number as integer.
• ValidationError: If port is non-numeric or out of range.

validate_port_range()

validate_port_range(range_str: str) -> list[int]

Validate a port range string and return the list of ports.

• range_str: Port range string (e.g., "8000-8010"). Returns: List of integer port numbers in the range [START, END].
• ValidationError: If format is invalid, ports are invalid,

validate_port_list()

validate_port_list(list_str: str) -> list[int]

Validate a comma/semicolon-separated port list.

• list_str: Comma or semicolon separated port numbers Returns: List of valid integer port numbers.
• ValidationError: If no valid ports are found in the list.

validate_hostname()

validate_hostname(host: str) -> str

Validate a hostname or IP address for use as a network target.

• host: Hostname or IP address string. Returns: The validated hostname string (unchanged).
• ValidationError: If hostname is empty, contains forbidden

is_ipv6_literal()

is_ipv6_literal(host: str) -> bool

Report whether a validated host is an IPv6 literal. Used to select an IPv6 connector for an IPv6 remote target. Hostnames and IPv4 literals return False.

validate_protocol()

validate_protocol(proto: str) -> str

Validate and normalize a protocol string.

• proto: Protocol string to validate. Returns: Normalized protocol string (tcp4, tcp6, udp4, or udp6).
• ValidationError: If protocol is not in the accepted whitelist.

validate_writable_path()

validate_writable_path(path: str) -> str

Validate a path that will be written to, such as a capture or log file. The file need not exist, so existence and readability are not required, but a non-empty path, no traversal component, and no forbidden characters are enforced using the shared config.FILEPATH_FORBIDDEN_CHARS set. validate_file_path() builds on this and adds the existence and readability checks for input files, so both validators reject the same character set from one definition.

validate_file_path()

validate_file_path(path: str) -> str

Validate a file path is safe for use.

• path: File path string to validate. Returns: The validated path string (unchanged).
• ValidationError: If path is empty, contains traversal sequences,

validate_socat_opts()

validate_socat_opts(opts: str) -> str

Validate user-provided socat address options for safety.

• opts: Socat options string to validate. Returns: The validated options string (unchanged).
• ValidationError: If options contain forbidden characters.

validate_session_name()

validate_session_name(name: str) -> str

Validate a user-provided session name for safety.

• name: Session name string to validate. Returns: The validated session name (unchanged).
• ValidationError: If name is empty, too long, or contains

validate_session_id()

validate_session_id(sid: str) -> str

Validate a session ID is a valid 8-character lowercase hex string.

• sid: Session ID string to validate. Returns: The validated session ID (unchanged).
• ValidationError: If session ID is empty or not exactly 8

---

Session Management Functions

Session files use KEY=VALUE text format. All functions use exact-key matching (line.split('=', 1)[0] == field) to prevent field confusion attacks (e.g., PID matching LAUNCHER_PID).

generate_session_id()

generate_session_id() -> str

Generate a unique 8-character hex session ID.

session_lock()

session_lock() -> Generator[None, None, None]

Acquire an advisory file lock on the session directory.

session_register()

session_register(sid: str, name: str, pid: int, pgid: int, mode: str, proto: str, lport: int, socat_cmd: str, rhost: str, rport: str) -> None

Register a new socat session by writing a session file.

session_update_process()

session_update_process(sid: str, pid: int, pgid: int) -> bool

Update the tracked process identity of an existing session.

Rewrites the PID and PGID fields of a session file and preserves every other field: session name, mode, protocol, ports, socat command, correlation ID, launcher PID, and the STARTED creation timestamp. STARTED is the session creation time and is not changed by a restart. Returns True when the record was updated, False when the session file does not exist or cannot be rewritten.

The session file is the authoritative record of which process a session owns. Any component that replaces the process behind a session calls this so that session_is_alive() and stop_session() act on the running process rather than a terminated predecessor. The watchdog calls it after every successful restart.

The read-modify-write cycle runs under the advisory session lock. The new record is written to a temporary file with 0o600 permissions and renamed over the original, which is atomic within a directory, so a concurrent reader never observes a partially written session record.

session_unregister()

session_unregister(sid: str) -> None

Remove a session file and all associated signal files.

session_read_field()

session_read_field(session_file: Path, field: str) -> str

Read a specific field from a session file using exact key match.

session_find_by_name()

session_find_by_name(target_name: str) -> list[str]

Find session IDs matching an exact session name.

session_find_by_port()

session_find_by_port(target_port: int | str) -> list[str]

Find session IDs matching a local port.

session_find_by_pid()

session_find_by_pid(target_pid: int | str) -> list[str]

Find session IDs matching a PID.

process_alive()

process_alive(pid_str: str, pgid_str: str) -> bool

Decide liveness from a recorded PID and PGID that the caller already holds. Checks the PID first, then the process group. The primary check routes through process_is_running(), so an exited-but-uncollected socat child is reported as dead rather than as a live zombie. The listing and detail views call this with the fields they have already read, so they do not re-open the session file for the alive check.

session_is_alive()

session_is_alive(sid: str) -> bool

Check if a registered session's process is still running.

session_get_all_ids()

session_get_all_ids() -> list[str]

List all registered session IDs.

session_count()

session_count() -> int

Count the number of active session files.

session_list()

session_list() -> bool

List all registered sessions with their status.

session_read_all_fields()

session_read_all_fields(session_file: Path) -> dict[str, str]

Read all fields from a session file in a single pass.

session_detail()

session_detail(sid: str) -> bool

Display detailed information for a specific session.

migrate_legacy_sessions()

migrate_legacy_sessions() -> int

Migrate v1 legacy .pid session files to v2.2+ .session format.

session_cleanup_dead()

session_cleanup_dead() -> int

Remove session files for dead processes.

---

Command Builder Functions

All builders produce list[str] argument lists (never shell strings). They perform no validation -- inputs are assumed sanitized by the trust boundary. Called by mode handlers after validation.

format_socat_host()

format_socat_host(host: str) -> str

Format a host for inclusion in a socat address.

socat address fields are colon-delimited, so an IPv6 literal is enclosed in square brackets to keep its own colons from being read as field separators. 2001:db8::1 becomes [2001:db8::1]. Hostnames and IPv4 literals contain no colons and pass through unchanged, and an already bracketed literal is returned as-is.

Without the brackets an address such as TCP6:2001:db8::1:443 is ambiguous -- the boundary between address and port cannot be determined -- and socat rejects it.

format_socat_endpoint()

format_socat_endpoint(host: str, port: int | str) -> str

Format a host and port as a socat endpoint in HOST:PORT form, bracketing IPv6 literals. The port remains the final colon-delimited field: [2001:db8::1]:443, 10.0.0.5:80.

cmd_list_to_string()

cmd_list_to_string(cmd: list[str]) -> str

Join a command argument list into a single string.

build_socat_listen_cmd()

build_socat_listen_cmd(proto: str, port: int, logfile: str, extra_opts: str, capture: bool) -> list[str]

Build a socat command for a single-port listener.

build_socat_forward_cmd()

build_socat_forward_cmd(listen_proto: str, lport: int, rhost: str, rport: int, remote_proto: str, capture: bool) -> list[str]

Build a socat command for bidirectional port forwarding.

build_socat_tunnel_cmd()

build_socat_tunnel_cmd(lport: int, rhost: str, rport: int, cert: str, key: str, capture: bool, remote_proto: str) -> list[str]

Build a socat command for an encrypted (OpenSSL) tunnel. The listener is TLS over TCP; the remote leg's transport is TCP and its address family follows remote_proto, so an IPv6 remote target is reached over a bracketed TCP6 connector. The tunnel mode derives this from the remote host, selecting TCP6 for an IPv6 literal and TCP4 otherwise.

Build a socat command for an encrypted (OpenSSL) tunnel.

build_socat_redirect_cmd()

build_socat_redirect_cmd(proto: str, lport: int, rhost: str, rport: int, capture: bool) -> list[str]

Build a socat command for transparent traffic redirection.

Capture Flag

When capture=True, builders insert -v into the socat argument list. This tells socat to dump all traffic in hex format to stderr. The launcher redirects stderr to a capture log file.

---

Process Management Functions

The most security-critical module. Controls process spawning (setsid + argument lists), termination (9-step protocol-scoped sequence), and port verification.

process_is_running()

process_is_running(pid: int) -> bool

Check whether a process is running, treating an exited but uncollected child as dead.

Processes launched by the framework are children of the management process. An exited child stays in the process table as a zombie until its exit status is collected, and a zombie still answers signal 0 -- so a check written against signal 0 alone would report a dead child as alive. For a child, the retained Popen handle is authoritative: polling it collects a terminated child and reports the truth. For any other process, signal 0 is used and qualified by a zombie check.

reap_child()

reap_child(pid: int) -> int | None

Collect the exit status of a terminated child and drop its handle, which removes the process table entry. Returns the exit status, or None if the PID is not a child of this process or is still running. A child that is still running is left alone.

register_child()

register_child(process: subprocess.Popen) -> None

Record a child process handle so its exit status can be collected when it terminates. Called by launch_socat_session() and by the watchdog on every replacement launch.

launch_socat_session()

launch_socat_session(name: str, mode: str, proto: str, lport: int, cmd: list[str], rhost: str, rport: str, stderr_redirect: str) -> tuple[str, int]

Launch a socat process with session tracking and process isolation.

check_port_available()

check_port_available(port: int, proto: str) -> bool

Check if a port is available for binding on a specific protocol.

The query is scoped to both the transport and the address family of the supplied protocol: a tcp4 check lists TCP listeners in the IPv4 family only. A tcp6 listener therefore never masks an available tcp4 port, and a UDP listener never masks an available TCP port. The two dimensions together are what make dual-stack operation correct.

check_port_freed()

check_port_freed(port: int, proto: str, retries: int) -> bool

Verify that a port has been released after stopping a session.

Retries with a delay to account for TIME_WAIT. Carries the same transport and address family scope as check_port_available(), so a listener of the other family on the same port number does not keep the port reporting as in use.

kill_by_port()

kill_by_port(port: int, proto: str) -> None

Last-resort function to kill socat processes on a specific port.

Enumeration is scoped to one transport and one address family, and every candidate PID is verified as socat through /proc/{pid}/comm before a signal is sent. Both scopes are required. A tcp4 listener and a tcp6 listener are independent sockets that can hold the same port number at the same time, so a query that named only the transport would enumerate the other family's socat process and terminate it -- stopping one session would take down an unrelated one.

stop_session()

stop_session(sid: str) -> bool

Execute the 9-step stop sequence for a session.

---

Watchdog Functions

Monitor-first auto-restart. The watchdog receives the PID of an already-running process and monitors it. It never launches the initial process -- only re-launches after confirmed death.

watchdog_loop()

watchdog_loop(session_id: str, session_name: str, cmd: list[str], initial_pid: int, max_restarts: int, backoff_initial: int, stderr_redirect: str) -> None

Background watchdog that monitors a process and auto-restarts on crash.

After each successful restart the watchdog calls session_update_process() with the replacement PID and PGID before it begins monitoring the new process. Under setsid the replacement is a session leader, so its PGID equals its PID. This keeps the session record aligned with the process that currently owns the port, which is what liveness reporting and the stop sequence depend on. A failed replacement launch leaves the record untouched and ends the watchdog.

start_watchdog()

start_watchdog(session_id: str, session_name: str, cmd: list[str], initial_pid: int, max_restarts: int, backoff_initial: int, stderr_redirect: str) -> threading.Thread

Start a watchdog monitor as a daemon thread.

---

Certificate Functions

generate_self_signed_cert()

generate_self_signed_cert(cn: str) -> tuple[str, str]

Generate a self-signed certificate and private key for TLS tunnels.

---

Adding a New Mode

To add a new operational mode (e.g., proxy):

1. Create src/socat_manager/modes/proxy.py with mode_proxy(args: Namespace) -> None
2. Follow the pattern: validate inputs → check port → build command → launch → watchdog → dual-stack
3. Add command builder to commands.py if the socat syntax differs from existing builders
4. Add subparser to cli.py in build_parser() with help text and examples
5. Add dispatch case to __main__.py in dispatch_mode()
6. Add menu submenu to menu.py
7. Add unit tests (test_cli parser, test_commands builder) and integration tests (test_mode_handlers)
8. Update documentation: README modes section, USAGE_GUIDE mode reference, wiki Usage-Guide

Code Style

• Type hints: All function signatures fully typed
• Docstrings: Google-style with Args, Returns, Raises sections
• Imports: from __future__ import annotations at top of every module
• Constants: UPPER_SNAKE_CASE, defined in config.py, Final[T] type hints
• Dataclasses: @dataclass(frozen=True, slots=True) for all configuration classes
• Line length: 120 characters (E501 ignored in ruff)
• Linting: ruff check --select E,W,F,I

Interactive Menu System

menu.py (948 lines, 28 functions) provides the full interactive TUI.

Key Functions

• interactive_menu() -- main loop with banner and 10-option menu
• _menu_listen() through _menu_redirect() -- per-mode submenus
• _collect_common_flags() -- shared protocol, dual-stack, capture, watchdog, name
• _confirm_and_execute() -- displays command, confirms, runs, catches SystemExit
• _offer_paired_forward() -- paired forward prompt after listener
• _prompt_port(), _prompt_host(), etc. -- validated input prompts

Cancel Convention

Every prompt function raises _MenuCancel on q, quit, cancel, back, exit, or Ctrl+C/Ctrl+D. Submenu functions catch _MenuCancel to return to the root menu. The main loop catches KeyboardInterrupt for graceful exit.

Exception Conventions

Exception	Raised By	Meaning
ValidationError	All validators	Invalid user input
RuntimeError	launch_socat_session()	Launch failed (port busy, socat error)
OSError	os.kill(), os.killpg()	Process/signal failure
SystemExit	Mode handlers (sys.exit(1))	Fatal validation or launch error
KeyboardInterrupt	User Ctrl+C	Interrupt signal
_MenuCancel	Menu prompt functions	User cancelled input

---

Socat Network Operations Manager · Repository · MIT License