mirror of
https://github.com/usestrix/strix.git
synced 2026-11-09 05:19:39 -10:00
Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b2bc4174bc | ||
|
|
6f3820f8eb | ||
|
|
4551ba95dc | ||
|
|
8f48276b33 |
No files matched your search
@@ -11,6 +11,7 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import unicodedata
|
||||
from datetime import UTC, datetime
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
@@ -58,6 +59,45 @@ class ModelConnectionError(RuntimeError):
|
||||
self.model_name = model_name
|
||||
|
||||
|
||||
def _first_non_ascii(value: str) -> tuple[int, str] | None:
|
||||
for position, char in enumerate(value):
|
||||
if ord(char) > 0x7F:
|
||||
return position, char
|
||||
return None
|
||||
|
||||
|
||||
def _header_candidates(
|
||||
prefix: str, model: str | None, api_key: str | None, extra_headers: dict[str, str] | None
|
||||
) -> list[tuple[str, str]]:
|
||||
candidates: list[tuple[str, str]] = []
|
||||
if api_key and not codex.subscription_model(model):
|
||||
candidates.append((f"{prefix}LLM_API_KEY", api_key))
|
||||
for header, value in (extra_headers or {}).items():
|
||||
candidates.append((f"{prefix}LLM_EXTRA_HEADERS header name {header!r}", header))
|
||||
candidates.append((f"{prefix}LLM_EXTRA_HEADERS value for {header!r}", value))
|
||||
return candidates
|
||||
|
||||
|
||||
def check_header_safe_credentials(settings: Settings) -> None:
|
||||
llm = settings.llm
|
||||
dedupe = settings.dedupe
|
||||
candidates = _header_candidates("", llm.model, llm.api_key, llm.extra_headers)
|
||||
if dedupe.model:
|
||||
candidates += _header_candidates(
|
||||
"DEDUPE_", dedupe.model, (dedupe.api_key or "").strip(), dedupe.extra_headers
|
||||
)
|
||||
for setting, value in candidates:
|
||||
found = _first_non_ascii(value)
|
||||
if found is None:
|
||||
continue
|
||||
position, char = found
|
||||
raise ValueError(
|
||||
f"{setting} contains a character that cannot be sent in an HTTP header: "
|
||||
f"U+{ord(char):04X} ({unicodedata.name(char, 'unnamed character')}) "
|
||||
f"at position {position + 1} of {len(value)}. Re-enter the value without it."
|
||||
)
|
||||
|
||||
|
||||
async def preflight_model_connection(
|
||||
model_name: str,
|
||||
*,
|
||||
@@ -70,6 +110,7 @@ async def preflight_model_connection(
|
||||
from strix.core.inputs import make_model_settings
|
||||
|
||||
resolved_settings = load_settings() if settings is None else settings
|
||||
check_header_safe_credentials(resolved_settings)
|
||||
configure_sdk_model_defaults(resolved_settings)
|
||||
model = StrixProvider().get_model(model_name)
|
||||
request_settings = make_model_settings(
|
||||
|
||||
@@ -0,0 +1,152 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
|
||||
import pytest
|
||||
|
||||
from strix.config import Settings
|
||||
from strix.interface.scan_setup import check_header_safe_credentials, preflight_model_connection
|
||||
|
||||
|
||||
_ENV_KEYS = (
|
||||
"LLM_API_KEY",
|
||||
"OPENAI_API_KEY",
|
||||
"LLM_EXTRA_HEADERS",
|
||||
"LLM_API_BASE",
|
||||
"STRIX_LLM",
|
||||
"STRIX_DEDUPE_MODEL",
|
||||
"DEDUPE_LLM_API_KEY",
|
||||
"DEDUPE_LLM_EXTRA_HEADERS",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clean_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
for key in _ENV_KEYS:
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
|
||||
|
||||
def _settings(monkeypatch: pytest.MonkeyPatch, **env: str) -> Settings:
|
||||
for key, value in env.items():
|
||||
monkeypatch.setenv(key, value)
|
||||
return Settings()
|
||||
|
||||
|
||||
def test_ascii_credentials_pass(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
settings = _settings(
|
||||
monkeypatch,
|
||||
LLM_API_KEY="sk-plain-ascii",
|
||||
LLM_EXTRA_HEADERS='{"X-Team": "security"}',
|
||||
)
|
||||
|
||||
check_header_safe_credentials(settings)
|
||||
|
||||
|
||||
def test_smart_quote_in_api_key_is_named_without_leaking_the_key(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
settings = _settings(monkeypatch, LLM_API_KEY="sk-abc\u201d")
|
||||
|
||||
with pytest.raises(ValueError, match="LLM_API_KEY") as excinfo:
|
||||
check_header_safe_credentials(settings)
|
||||
|
||||
message = str(excinfo.value)
|
||||
assert "U+201D" in message
|
||||
assert "RIGHT DOUBLE QUOTATION MARK" in message
|
||||
assert "position 7 of 7" in message
|
||||
assert "sk-abc" not in message
|
||||
|
||||
|
||||
@pytest.mark.parametrize("char", ["\u00a0", "\ufeff", "\u200b"])
|
||||
def test_invisible_characters_are_reported(monkeypatch: pytest.MonkeyPatch, char: str) -> None:
|
||||
settings = _settings(monkeypatch, LLM_API_KEY=f"{char}sk-abc")
|
||||
|
||||
with pytest.raises(ValueError, match=f"U\\+{ord(char):04X}"):
|
||||
check_header_safe_credentials(settings)
|
||||
|
||||
|
||||
def test_extra_header_value_is_checked(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
settings = _settings(
|
||||
monkeypatch,
|
||||
LLM_API_KEY="sk-plain",
|
||||
LLM_EXTRA_HEADERS='{"X-Team": "s\u00e9curit\u00e9"}',
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match=r"LLM_EXTRA_HEADERS value for 'X-Team'.*U\+00E9"):
|
||||
check_header_safe_credentials(settings)
|
||||
|
||||
|
||||
def test_subscription_model_ignores_an_unused_main_key(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
settings = _settings(monkeypatch, STRIX_LLM="chatgpt/gpt-5", LLM_API_KEY="sk-abc\u201d")
|
||||
|
||||
check_header_safe_credentials(settings)
|
||||
|
||||
|
||||
def test_subscription_model_still_checks_extra_headers(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
settings = _settings(
|
||||
monkeypatch,
|
||||
STRIX_LLM="chatgpt/gpt-5",
|
||||
LLM_EXTRA_HEADERS='{"X-Team": "s\u00e9curit\u00e9"}',
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match=r"LLM_EXTRA_HEADERS value for 'X-Team'.*U\+00E9"):
|
||||
check_header_safe_credentials(settings)
|
||||
|
||||
|
||||
def test_dedupe_key_is_checked_when_a_dedupe_model_is_set(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
settings = _settings(
|
||||
monkeypatch,
|
||||
LLM_API_KEY="sk-plain",
|
||||
STRIX_DEDUPE_MODEL="openai/gpt-4o-mini",
|
||||
DEDUPE_LLM_API_KEY="sk-de\u00a0dupe",
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match=r"DEDUPE_LLM_API_KEY.*U\+00A0.*position 6 of 10"):
|
||||
check_header_safe_credentials(settings)
|
||||
|
||||
|
||||
def test_dedupe_key_is_checked_as_sent_after_resolve_dedupe_model_strips_it(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
settings = _settings(
|
||||
monkeypatch,
|
||||
LLM_API_KEY="sk-plain",
|
||||
STRIX_DEDUPE_MODEL="openai/gpt-4o-mini",
|
||||
DEDUPE_LLM_API_KEY="\u00a0sk-dedupe\u00a0",
|
||||
)
|
||||
|
||||
check_header_safe_credentials(settings)
|
||||
|
||||
|
||||
def test_dedupe_subscription_model_checks_headers_but_not_the_key(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
settings = _settings(
|
||||
monkeypatch,
|
||||
LLM_API_KEY="sk-plain",
|
||||
STRIX_DEDUPE_MODEL="chatgpt/gpt-5",
|
||||
DEDUPE_LLM_API_KEY="sk-dedupe\u201d",
|
||||
DEDUPE_LLM_EXTRA_HEADERS='{"X-Team": "s\u00e9curit\u00e9"}',
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match=r"DEDUPE_LLM_EXTRA_HEADERS value for 'X-Team'"):
|
||||
check_header_safe_credentials(settings)
|
||||
|
||||
|
||||
def test_dedupe_key_is_ignored_without_a_dedupe_model(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
settings = _settings(monkeypatch, LLM_API_KEY="sk-plain", DEDUPE_LLM_API_KEY="sk-dedupe\u00a0")
|
||||
|
||||
check_header_safe_credentials(settings)
|
||||
|
||||
|
||||
def test_preflight_rejects_the_key_before_any_request(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
settings = _settings(
|
||||
monkeypatch,
|
||||
LLM_API_KEY="sk-abc\u201d",
|
||||
LLM_API_BASE="http://127.0.0.1:9/v1",
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match="LLM_API_KEY"):
|
||||
asyncio.run(preflight_model_connection("openai/gpt-4o-mini", settings=settings))
|
||||
Reference in new issue
Block a user