Skip to content

Commit 6bfed5c

Browse files
committed
Extract shared loopback guard and fix small lint nits
Remove the duplicated _ensure_loopback copies in http_server.py and tcp_server.py in favour of a new server/network_guards.ensure_loopback helper. Rename log_message's `format` parameter to `format_str` to stop shadowing the builtin, and rename the unused `args` in _cmd_ui to _args so lint stops flagging it.
1 parent d775019 commit 6bfed5c

4 files changed

Lines changed: 33 additions & 37 deletions

File tree

automation_file/__main__.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -98,7 +98,7 @@ def _cmd_http_server(args: argparse.Namespace) -> int:
9898
return 0
9999

100100

101-
def _cmd_ui(args: argparse.Namespace) -> int:
101+
def _cmd_ui(_args: argparse.Namespace) -> int:
102102
from automation_file.ui.launcher import launch_ui
103103

104104
return launch_ui()

automation_file/server/http_server.py

Lines changed: 4 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -11,16 +11,15 @@
1111
from __future__ import annotations
1212

1313
import hmac
14-
import ipaddress
1514
import json
16-
import socket
1715
import threading
1816
from http import HTTPStatus
1917
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
2018

2119
from automation_file.core.action_executor import execute_action
2220
from automation_file.exceptions import TCPAuthException
2321
from automation_file.logging_config import file_automation_logger
22+
from automation_file.server.network_guards import ensure_loopback
2423

2524
_DEFAULT_HOST = "127.0.0.1"
2625
_DEFAULT_PORT = 9944
@@ -30,8 +29,8 @@
3029
class _HTTPActionHandler(BaseHTTPRequestHandler):
3130
"""POST /actions -> JSON results."""
3231

33-
def log_message(self, format: str, *args: object) -> None:
34-
file_automation_logger.info("http_server: " + format, *args)
32+
def log_message(self, format_str: str, *args: object) -> None:
33+
file_automation_logger.info("http_server: " + format_str, *args)
3534

3635
def do_POST(self) -> None:
3736
if self.path != "/actions":
@@ -100,20 +99,6 @@ def __init__(
10099
self.shared_secret: str | None = shared_secret
101100

102101

103-
def _ensure_loopback(host: str) -> None:
104-
try:
105-
infos = socket.getaddrinfo(host, None)
106-
except socket.gaierror as error:
107-
raise ValueError(f"cannot resolve host: {host}") from error
108-
for info in infos:
109-
ip_obj = ipaddress.ip_address(info[4][0])
110-
if not ip_obj.is_loopback:
111-
raise ValueError(
112-
f"host {host} resolves to non-loopback {ip_obj}; pass allow_non_loopback=True "
113-
"if exposure is intentional"
114-
)
115-
116-
117102
def start_http_action_server(
118103
host: str = _DEFAULT_HOST,
119104
port: int = _DEFAULT_PORT,
@@ -122,7 +107,7 @@ def start_http_action_server(
122107
) -> HTTPActionServer:
123108
"""Start the HTTP action server on a background thread."""
124109
if not allow_non_loopback:
125-
_ensure_loopback(host)
110+
ensure_loopback(host)
126111
if allow_non_loopback and not shared_secret:
127112
file_automation_logger.warning(
128113
"http_server: non-loopback bind without shared_secret is insecure",
Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
"""Network-binding guards shared by every embedded action server."""
2+
3+
from __future__ import annotations
4+
5+
import ipaddress
6+
import socket
7+
8+
9+
def ensure_loopback(host: str) -> None:
10+
"""Raise ``ValueError`` if ``host`` resolves to a non-loopback address.
11+
12+
Every resolved A / AAAA record must be loopback. The explicit error message
13+
names the opt-out flag so callers are reminded that exposing a server
14+
dispatching arbitrary registry commands is equivalent to a remote REPL.
15+
"""
16+
try:
17+
infos = socket.getaddrinfo(host, None)
18+
except socket.gaierror as error:
19+
raise ValueError(f"cannot resolve host: {host}") from error
20+
for info in infos:
21+
ip_obj = ipaddress.ip_address(info[4][0])
22+
if not ip_obj.is_loopback:
23+
raise ValueError(
24+
f"host {host} resolves to non-loopback {ip_obj}; pass allow_non_loopback=True "
25+
"if exposure is intentional"
26+
)

automation_file/server/tcp_server.py

Lines changed: 2 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -13,9 +13,7 @@
1313
from __future__ import annotations
1414

1515
import hmac
16-
import ipaddress
1716
import json
18-
import socket
1917
import socketserver
2018
import sys
2119
import threading
@@ -24,6 +22,7 @@
2422
from automation_file.core.action_executor import execute_action
2523
from automation_file.exceptions import TCPAuthException
2624
from automation_file.logging_config import file_automation_logger
25+
from automation_file.server.network_guards import ensure_loopback
2726

2827
_DEFAULT_HOST = "localhost"
2928
_DEFAULT_PORT = 9943
@@ -116,20 +115,6 @@ def __init__(
116115
self.shared_secret: str | None = shared_secret
117116

118117

119-
def _ensure_loopback(host: str) -> None:
120-
try:
121-
infos = socket.getaddrinfo(host, None)
122-
except socket.gaierror as error:
123-
raise ValueError(f"cannot resolve host: {host}") from error
124-
for info in infos:
125-
ip_obj = ipaddress.ip_address(info[4][0])
126-
if not ip_obj.is_loopback:
127-
raise ValueError(
128-
f"host {host} resolves to non-loopback {ip_obj}; pass allow_non_loopback=True "
129-
"if exposure is intentional"
130-
)
131-
132-
133118
def start_autocontrol_socket_server(
134119
host: str = _DEFAULT_HOST,
135120
port: int = _DEFAULT_PORT,
@@ -143,7 +128,7 @@ def start_autocontrol_socket_server(
143128
address without a shared secret is strongly discouraged.
144129
"""
145130
if not allow_non_loopback:
146-
_ensure_loopback(host)
131+
ensure_loopback(host)
147132
if allow_non_loopback and not shared_secret:
148133
file_automation_logger.warning(
149134
"tcp_server: non-loopback bind without shared_secret is insecure",

0 commit comments

Comments
 (0)