Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
148 changes: 147 additions & 1 deletion src/qwenpaw/cli/app_cmd.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,151 @@
# -*- coding: utf-8 -*-
from __future__ import annotations

import ipaddress
import logging
import os
import socket
import sys

import click
import uvicorn

from ..constant import LOG_LEVEL_ENV
from ..constant import LOG_LEVEL_ENV, EnvVarLoader
from ..config.utils import write_last_api
from ..utils.logging import setup_logger, SuppressPathAccessLogFilter


_LOOPBACK_HOSTNAMES = frozenset(
{"localhost", "localhost.localdomain", "ip6-localhost"},
)


def _addr_is_loopback(addr: str) -> bool:
try:
return ipaddress.ip_address(addr).is_loopback
except ValueError:
return False


def _host_is_loopback(host: str) -> bool:
"""Return True when *host* binds only to a loopback interface.

Accepts IPv4/IPv6 literals and a small allowlist of well-known
loopback hostnames. Anything that resolves to a non-loopback
address — including ``0.0.0.0``, ``::``, and arbitrary public
interfaces or hostnames — returns False.
"""
if not host:
return False
if host.lower() in _LOOPBACK_HOSTNAMES:
return True
if _addr_is_loopback(host):
return True
# Hostname: only treat as loopback if every resolved address is loopback.
try:
infos = socket.getaddrinfo(host, None)
except socket.gaierror:
return False
return bool(infos) and all(_addr_is_loopback(info[4][0]) for info in infos)


def _enforce_unauth_public_bind_safety(
host: str,
allow_unauth_public: bool,
) -> None:
"""Refuse non-loopback bind when no auth gate is configured.

QwenPaw's HTTP gateway can invoke host-affecting tools (shell
commands, file IO, etc.). Authentication is opt-in via
``QWENPAW_AUTH_ENABLED``; binding to a non-loopback address with
auth disabled effectively exposes a tool-enabled agent on the
network with no gate. We refuse this configuration unless the
operator explicitly opts in via ``--allow-unauth-public`` or the
``QWENPAW_ALLOW_UNAUTH_PUBLIC`` env var.
"""
if _host_is_loopback(host):
return

# Lazy import to avoid pulling app stack into CLI startup.
from ..app.auth import is_auth_enabled

if is_auth_enabled():
return

env_override = EnvVarLoader.get_str(
"QWENPAW_ALLOW_UNAUTH_PUBLIC",
"",
).strip().lower() in ("true", "1", "yes")
if allow_unauth_public or env_override:
click.echo(
"⚠️ WARNING: binding to a non-loopback address "
f"({host}) with authentication disabled.",
err=True,
)
click.echo(
" The QwenPaw HTTP gateway exposes tool-enabled agents that "
"can run shell commands, read files, and call external APIs.",
err=True,
)
click.echo(
" Anyone who can reach this port can drive those tools. "
"Make sure auth is enforced upstream (reverse proxy, VPN, "
"Tailscale, security group) before exposing this port.",
err=True,
)
click.echo(err=True)
return

click.echo(
f"❌ Refusing to bind to non-loopback host '{host}' with "
"authentication disabled.",
err=True,
)
click.echo(err=True)
click.echo(
"QwenPaw's HTTP gateway can invoke host-affecting tools (shell "
"commands, file IO, external APIs). Exposing it on a non-loopback "
"interface without an authentication gate would let anyone who "
"reaches the port drive those tools.",
err=True,
)
click.echo(err=True)
click.echo("To proceed, choose one of the following:", err=True)
click.echo(err=True)
click.echo(
" 1. (recommended) Bind to loopback and put a reverse proxy / "
"Tailscale / VPN in front:",
err=True,
)
click.echo(" qwenpaw app --host 127.0.0.1 --port <PORT>", err=True)
click.echo(err=True)
click.echo(" 2. Enable QwenPaw's built-in authentication:", err=True)
click.echo(" export QWENPAW_AUTH_ENABLED=true", err=True)
click.echo(
" qwenpaw app --host <HOST> --port <PORT> "
"# then register at /login",
err=True,
)
click.echo(err=True)
click.echo(
" 3. Override the safety check (only if auth is enforced "
"upstream by a reverse proxy / VPN you control):",
err=True,
)
click.echo(
" qwenpaw app --host <HOST> --port <PORT> "
"--allow-unauth-public",
err=True,
)
click.echo(
" # or set QWENPAW_ALLOW_UNAUTH_PUBLIC=true in the service "
"environment.",
err=True,
)
click.echo(err=True)
sys.exit(2)


@click.command("app")
@click.option(
"--host",
Expand Down Expand Up @@ -52,13 +186,23 @@
"This option is deprecated and will be removed in a future version. "
"QwenPaw always uses 1 worker.",
)
@click.option(
"--allow-unauth-public",
is_flag=True,
default=False,
help="Allow binding to a non-loopback address even when "
"QWENPAW_AUTH_ENABLED is not set. Only use this when an upstream "
"reverse proxy or VPN enforces authentication. Can also be set via "
"the QWENPAW_ALLOW_UNAUTH_PUBLIC environment variable.",
)
def app_cmd(
host: str,
port: int,
reload: bool,
workers: int, # pylint: disable=unused-argument
log_level: str,
hide_access_paths: tuple[str, ...],
allow_unauth_public: bool,
) -> None:
"""Run QwenPaw FastAPI app."""
# Handle deprecated --workers parameter
Expand All @@ -75,6 +219,8 @@ def app_cmd(
)
click.echo(err=True)

_enforce_unauth_public_bind_safety(host, allow_unauth_public)

# Persist last used host/port for other terminals
if host == "0.0.0.0":
write_last_api("127.0.0.1", port)
Expand Down
182 changes: 182 additions & 0 deletions tests/unit/cli/test_cli_app_safety.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,182 @@
# -*- coding: utf-8 -*-
"""Tests for the unauthenticated public-bind safety gate in ``qwenpaw app``.

QwenPaw's HTTP gateway can invoke host-affecting tools. When
``QWENPAW_AUTH_ENABLED`` is unset, binding to a non-loopback host
exposes those tools without an authentication gate. ``app_cmd`` must
refuse this configuration unless the operator opts in.
"""
# pylint: disable=protected-access,redefined-outer-name

from __future__ import annotations

from unittest.mock import patch

import pytest
from click.testing import CliRunner

from qwenpaw.cli import app_cmd as app_cmd_mod


@pytest.fixture
def runner() -> CliRunner:
return CliRunner()


@pytest.fixture(autouse=True)
def _isolate_env(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("QWENPAW_AUTH_ENABLED", raising=False)
monkeypatch.delenv("QWENPAW_ALLOW_UNAUTH_PUBLIC", raising=False)


# ---------------------------------------------------------------------------
# _host_is_loopback
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
"host",
["127.0.0.1", "127.5.5.5", "::1", "localhost", "LOCALHOST"],
)
def test_host_is_loopback_true(host: str) -> None:
assert app_cmd_mod._host_is_loopback(host) is True


@pytest.mark.parametrize(
"host",
["0.0.0.0", "::", "1.2.3.4", "192.168.1.10", "example.com", ""],
)
def test_host_is_loopback_false(host: str) -> None:
assert app_cmd_mod._host_is_loopback(host) is False


# ---------------------------------------------------------------------------
# _enforce_unauth_public_bind_safety
# ---------------------------------------------------------------------------


def _patch_auth(enabled: bool):
return patch.object(
app_cmd_mod,
"_enforce_unauth_public_bind_safety",
wraps=app_cmd_mod._enforce_unauth_public_bind_safety,
), patch("qwenpaw.app.auth.is_auth_enabled", return_value=enabled)


def test_loopback_bind_passes_without_auth() -> None:
with patch("qwenpaw.app.auth.is_auth_enabled", return_value=False):
# Should not raise / exit.
app_cmd_mod._enforce_unauth_public_bind_safety("127.0.0.1", False)


def test_public_bind_with_auth_enabled_passes() -> None:
with patch("qwenpaw.app.auth.is_auth_enabled", return_value=True):
app_cmd_mod._enforce_unauth_public_bind_safety("0.0.0.0", False)


def test_public_bind_with_explicit_flag_passes() -> None:
with patch("qwenpaw.app.auth.is_auth_enabled", return_value=False):
app_cmd_mod._enforce_unauth_public_bind_safety("0.0.0.0", True)


def test_public_bind_with_env_override_passes(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("QWENPAW_ALLOW_UNAUTH_PUBLIC", "true")
with patch("qwenpaw.app.auth.is_auth_enabled", return_value=False):
app_cmd_mod._enforce_unauth_public_bind_safety("0.0.0.0", False)


def test_public_bind_without_auth_or_override_exits() -> None:
with patch("qwenpaw.app.auth.is_auth_enabled", return_value=False):
with pytest.raises(SystemExit) as excinfo:
app_cmd_mod._enforce_unauth_public_bind_safety("0.0.0.0", False)
assert excinfo.value.code == 2


# ---------------------------------------------------------------------------
# CLI integration
# ---------------------------------------------------------------------------


def test_app_cmd_refuses_public_bind_without_auth(runner: CliRunner) -> None:
with (
patch("qwenpaw.app.auth.is_auth_enabled", return_value=False),
patch(
"uvicorn.run",
) as mock_run,
):
result = runner.invoke(
app_cmd_mod.app_cmd,
["--host", "0.0.0.0", "--port", "8088"],
)
assert result.exit_code == 2
assert "Refusing to bind" in result.output
assert "QWENPAW_AUTH_ENABLED" in result.output
assert "--allow-unauth-public" in result.output
mock_run.assert_not_called()


def test_app_cmd_loopback_default_runs(runner: CliRunner) -> None:
with (
patch("qwenpaw.app.auth.is_auth_enabled", return_value=False),
patch(
"uvicorn.run",
) as mock_run,
patch("qwenpaw.cli.app_cmd.write_last_api"),
patch(
"qwenpaw.cli.app_cmd.setup_logger",
),
):
result = runner.invoke(
app_cmd_mod.app_cmd,
["--host", "127.0.0.1", "--port", "8088"],
)
assert result.exit_code == 0, result.output
mock_run.assert_called_once()


def test_app_cmd_public_bind_with_flag_runs(runner: CliRunner) -> None:
with (
patch("qwenpaw.app.auth.is_auth_enabled", return_value=False),
patch(
"uvicorn.run",
) as mock_run,
patch("qwenpaw.cli.app_cmd.write_last_api"),
patch(
"qwenpaw.cli.app_cmd.setup_logger",
),
):
result = runner.invoke(
app_cmd_mod.app_cmd,
[
"--host",
"0.0.0.0",
"--port",
"8088",
"--allow-unauth-public",
],
)
assert result.exit_code == 0, result.output
assert "WARNING" in result.output
mock_run.assert_called_once()


def test_app_cmd_public_bind_with_auth_env_runs(
runner: CliRunner,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("QWENPAW_AUTH_ENABLED", "true")
with (
patch("uvicorn.run") as mock_run,
patch(
"qwenpaw.cli.app_cmd.write_last_api",
),
patch("qwenpaw.cli.app_cmd.setup_logger"),
):
result = runner.invoke(
app_cmd_mod.app_cmd,
["--host", "0.0.0.0", "--port", "8088"],
)
assert result.exit_code == 0, result.output
mock_run.assert_called_once()