refactor(free-tier): keep auth.py and anon_challenge.py inside the code-health ratchet

- _nous_portal_base_url moves from the auth.py facade to auth_nous.py
  (its sibling _nous_portal_env_override already lives there) and is
  re-exported with the rest of the Nous names, offsetting the
  run_with_challenge seam: auth.py ends below its main value.
- _announce no longer wraps the gateway broadcast in a blind
  `except Exception`: _broadcast_global_event already contains a wedged
  peer per transport, so anything else is a bug worth surfacing.
- _present_in_terminal catches what webbrowser.open can raise
  (webbrowser.Error, OSError) instead of Exception.

No behaviour change.
This commit is contained in:
kshitijk4poor
2026-10-06 17:51:00 +05:30
committed by kshitij
parent 8c0568ee73
commit 9cb52daec2
3 changed files with 27 additions and 29 deletions

View File

@@ -324,12 +324,9 @@ def _announce(challenge: BrowserChallenge) -> bool:
server = sys.modules.get("tui_gateway.server")
if server is None:
return False
try:
server._broadcast_global_event(CHALLENGE_EVENT, challenge.as_payload())
return True
except Exception as exc:
logger.debug("%s not broadcast: %s", CHALLENGE_EVENT, exc)
return False
# The broadcast contains a wedged peer itself (per-transport); anything else is a real bug.
server._broadcast_global_event(CHALLENGE_EVENT, challenge.as_payload())
return True
def _present_in_terminal(challenge: BrowserChallenge) -> None:
@@ -340,13 +337,13 @@ def _present_in_terminal(challenge: BrowserChallenge) -> None:
already_open = challenge.url in _opened_urls
if (not already_open and client_surface() != "gateway"
and not _is_remote_session() and _can_open_graphical_browser()):
import webbrowser
try:
import webbrowser
opened = bool(webbrowser.open(challenge.url))
if opened:
_opened_urls.add(challenge.url)
except Exception as exc:
except (webbrowser.Error, OSError) as exc:
logger.debug("could not open the challenge in a browser: %s", exc)
if opened:
_opened_urls.add(challenge.url)
print(f"\n{challenge.message}", file=sys.stderr)
print(f" Open: {challenge.url}", file=sys.stderr)
print(" (Opened in your browser.)\n" if opened or already_open

View File

@@ -54,7 +54,7 @@ from hermes_cli.auth_nous import ( # noqa: F401 re-exported
_compute_nous_auth_status, _format_nous_entitlement_auth_error, _healed_nous_inference_url,
_login_nous, _merge_shared_nous_oauth_state, _migrate_stale_nous_portal_url,
_nous_device_code_login, _nous_inference_env_override, _nous_invoke_jwt_is_usable,
_nous_invoke_jwt_status, _nous_portal_env_override, _nous_shared_store_lock,
_nous_invoke_jwt_status, _nous_portal_base_url, _nous_portal_env_override, _nous_shared_store_lock,
_nous_shared_store_path, _pool_first_oauth_status, _quarantine_nous_oauth_state,
_quarantine_nous_pool_entries, _read_shared_nous_state, _refresh_access_token,
_refresh_nous_or_quarantine, _select_nous_invoke_jwt, _sync_nous_pool_from_auth_store,
@@ -1806,24 +1806,6 @@ _RESOLVE_TOKEN_CACHE: "dict[str, tuple[float, str]]" = {}
_RESOLVE_TOKEN_CACHE_TTL_S = 5.0
def _nous_portal_base_url(state: Dict[str, Any]) -> str:
"""HERMES_PORTAL_BASE_URL / NOUS_PORTAL_BASE_URL is the trusted operator override and wins
OUTRIGHT, bypassing the host allowlist (which exists to reject an untrusted network-provided
value, not one the operator configured). Otherwise the stored/default value, allowlist-gated."""
env_portal_override = _nous_portal_env_override()
if env_portal_override:
return env_portal_override.rstrip("/")
portal_base_url = _optional_base_url(state.get("portal_base_url")) or DEFAULT_NOUS_PORTAL_URL
portal_base_url = portal_base_url.rstrip("/")
host = urlparse(portal_base_url).hostname
if host and host not in _NOUS_PORTAL_ALLOWED_HOSTS:
logger.warning(
"auth: ignoring invalid portal_base_url %r (host %r not in allowlist), using default",
portal_base_url, host)
return DEFAULT_NOUS_PORTAL_URL
return portal_base_url
def resolve_nous_access_token(
*,
timeout_seconds: float = 15.0,

View File

@@ -212,6 +212,25 @@ def _nous_portal_env_override() -> Optional[str]:
return _optional_base_url(_scoped_operator_override("HERMES_PORTAL_BASE_URL", "NOUS_PORTAL_BASE_URL"))
def _nous_portal_base_url(state: Dict[str, Any]) -> str:
"""HERMES_PORTAL_BASE_URL / NOUS_PORTAL_BASE_URL is the trusted operator override and wins
OUTRIGHT, bypassing the host allowlist (which exists to reject an untrusted network-provided
value, not one the operator configured). Otherwise the stored/default value, allowlist-gated."""
env_portal_override = _nous_portal_env_override()
if env_portal_override:
return env_portal_override.rstrip("/")
from hermes_cli.auth import _NOUS_PORTAL_ALLOWED_HOSTS, _optional_base_url
portal_base_url = _optional_base_url(state.get("portal_base_url")) or DEFAULT_NOUS_PORTAL_URL
portal_base_url = portal_base_url.rstrip("/")
host = urlparse(portal_base_url).hostname
if host and host not in _NOUS_PORTAL_ALLOWED_HOSTS:
logger.warning(
"auth: ignoring invalid portal_base_url %r (host %r not in allowlist), using default",
portal_base_url, host)
return DEFAULT_NOUS_PORTAL_URL
return portal_base_url
def _scope_values(raw_scope: Any) -> set[str]:
# OAuth token responses return a space-separated string; collections are kept for JWT ``scp``
# claims and older stored fixtures.