mirror of
https://github.com/usestrix/strix.git
synced 2026-08-20 18:38:57 +02:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ac0014fe65 | ||
|
|
86282e83a8 |
+1
-1
@@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "strix-agent"
|
name = "strix-agent"
|
||||||
version = "1.4.0"
|
version = "1.4.1"
|
||||||
description = "Open-source AI Hackers for your apps"
|
description = "Open-source AI Hackers for your apps"
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
license = "Apache-2.0"
|
license = "Apache-2.0"
|
||||||
|
|||||||
+13
-20
@@ -18,12 +18,12 @@ import logging
|
|||||||
import secrets
|
import secrets
|
||||||
import threading
|
import threading
|
||||||
import time
|
import time
|
||||||
import urllib.error
|
|
||||||
import urllib.parse
|
import urllib.parse
|
||||||
import urllib.request
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import TYPE_CHECKING, Any
|
from typing import TYPE_CHECKING, Any
|
||||||
|
|
||||||
|
import requests
|
||||||
|
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from collections.abc import Iterator
|
from collections.abc import Iterator
|
||||||
@@ -221,26 +221,19 @@ def _first(query: dict[str, list[str]], key: str) -> str | None:
|
|||||||
|
|
||||||
|
|
||||||
def _post_form(payload: dict[str, str]) -> dict[str, Any]:
|
def _post_form(payload: dict[str, str]) -> dict[str, Any]:
|
||||||
body = urllib.parse.urlencode(payload).encode("ascii")
|
|
||||||
request = urllib.request.Request( # noqa: S310 - fixed https OAuth endpoint
|
|
||||||
TOKEN_URL,
|
|
||||||
data=body,
|
|
||||||
headers={
|
|
||||||
"Content-Type": "application/x-www-form-urlencoded",
|
|
||||||
"Accept": "application/json",
|
|
||||||
},
|
|
||||||
method="POST",
|
|
||||||
)
|
|
||||||
try:
|
try:
|
||||||
with urllib.request.urlopen( # noqa: S310 # nosec B310 - fixed https endpoint
|
response = requests.post(
|
||||||
request, timeout=_TOKEN_TIMEOUT
|
TOKEN_URL,
|
||||||
) as response:
|
data=payload,
|
||||||
data = json.loads(response.read() or b"{}")
|
headers={"Accept": "application/json"},
|
||||||
except urllib.error.HTTPError as exc:
|
timeout=_TOKEN_TIMEOUT,
|
||||||
detail = exc.read().decode("utf-8", "replace")[:300]
|
)
|
||||||
raise CodexAuthError("token_http_error", f"HTTP {exc.code}: {detail}") from exc
|
except requests.RequestException as exc:
|
||||||
except (urllib.error.URLError, TimeoutError, OSError) as exc:
|
|
||||||
raise CodexAuthError("unavailable", str(exc)) from exc
|
raise CodexAuthError("unavailable", str(exc)) from exc
|
||||||
|
if response.status_code >= 400:
|
||||||
|
detail = response.text[:300]
|
||||||
|
raise CodexAuthError("token_http_error", f"HTTP {response.status_code}: {detail}")
|
||||||
|
data = json.loads(response.content or b"{}")
|
||||||
if not isinstance(data, dict):
|
if not isinstance(data, dict):
|
||||||
raise CodexAuthError("bad_response", "token endpoint returned non-object")
|
raise CodexAuthError("bad_response", "token endpoint returned non-object")
|
||||||
return data
|
return data
|
||||||
|
|||||||
@@ -11,11 +11,10 @@ import tempfile
|
|||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any
|
from typing import Any
|
||||||
from urllib.error import HTTPError, URLError
|
|
||||||
from urllib.parse import urlparse
|
from urllib.parse import urlparse
|
||||||
from urllib.request import Request, urlopen
|
|
||||||
|
|
||||||
import docker
|
import docker
|
||||||
|
import requests
|
||||||
from docker.errors import DockerException, ImageNotFound
|
from docker.errors import DockerException, ImageNotFound
|
||||||
from rich.console import Console
|
from rich.console import Console
|
||||||
from rich.panel import Panel
|
from rich.panel import Panel
|
||||||
@@ -1088,13 +1087,12 @@ def resolve_diff_scope_context(
|
|||||||
def _is_http_git_repo(url: str) -> bool:
|
def _is_http_git_repo(url: str) -> bool:
|
||||||
check_url = f"{url.rstrip('/')}/info/refs?service=git-upload-pack"
|
check_url = f"{url.rstrip('/')}/info/refs?service=git-upload-pack"
|
||||||
try:
|
try:
|
||||||
req = Request(check_url, headers={"User-Agent": "git/strix"}) # noqa: S310
|
resp = requests.get(check_url, headers={"User-Agent": "git/strix"}, timeout=10)
|
||||||
with urlopen(req, timeout=10) as resp: # noqa: S310 # nosec B310
|
except (requests.RequestException, ValueError):
|
||||||
return "x-git-upload-pack-advertisement" in resp.headers.get("Content-Type", "")
|
|
||||||
except HTTPError as e:
|
|
||||||
return e.code == 401
|
|
||||||
except (URLError, OSError, ValueError):
|
|
||||||
return False
|
return False
|
||||||
|
if resp.status_code >= 400:
|
||||||
|
return resp.status_code == 401
|
||||||
|
return "x-git-upload-pack-advertisement" in resp.headers.get("Content-Type", "")
|
||||||
|
|
||||||
|
|
||||||
def infer_target_type(target: str) -> tuple[str, dict[str, str]]: # noqa: PLR0911
|
def infer_target_type(target: str) -> tuple[str, dict[str, str]]: # noqa: PLR0911
|
||||||
|
|||||||
@@ -15,12 +15,12 @@ import base64
|
|||||||
import contextlib
|
import contextlib
|
||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
import urllib.error
|
|
||||||
import urllib.request
|
|
||||||
from datetime import UTC, datetime
|
from datetime import UTC, datetime
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
|
import requests
|
||||||
|
|
||||||
from strix.config.loader import load_settings
|
from strix.config.loader import load_settings
|
||||||
|
|
||||||
|
|
||||||
@@ -147,21 +147,17 @@ def _post_json(path: str, payload: dict[str, Any], *, timeout: int) -> tuple[int
|
|||||||
map, not raised.
|
map, not raised.
|
||||||
"""
|
"""
|
||||||
url = f"{_app_url()}{path}"
|
url = f"{_app_url()}{path}"
|
||||||
body = json.dumps(payload).encode("utf-8")
|
|
||||||
request = urllib.request.Request( # noqa: S310 - fixed https relay URL
|
|
||||||
url,
|
|
||||||
data=body,
|
|
||||||
headers={"Content-Type": "application/json", "Accept": "application/json"},
|
|
||||||
method="POST",
|
|
||||||
)
|
|
||||||
try:
|
try:
|
||||||
with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 # nosec B310
|
response = requests.post(
|
||||||
return response.status, _parse_body(response.read())
|
url,
|
||||||
except urllib.error.HTTPError as exc:
|
json=payload,
|
||||||
return exc.code, _parse_body(exc.read())
|
headers={"Accept": "application/json"},
|
||||||
except (urllib.error.URLError, TimeoutError, OSError) as exc:
|
timeout=timeout,
|
||||||
|
)
|
||||||
|
except requests.RequestException as exc:
|
||||||
logger.warning("relay request to %s failed: %s", path, exc)
|
logger.warning("relay request to %s failed: %s", path, exc)
|
||||||
raise RelayError("unavailable") from exc
|
raise RelayError("unavailable") from exc
|
||||||
|
return response.status_code, _parse_body(response.content)
|
||||||
|
|
||||||
|
|
||||||
def _parse_body(raw: bytes) -> dict[str, Any]:
|
def _parse_body(raw: bytes) -> dict[str, Any]:
|
||||||
|
|||||||
@@ -1,9 +1,9 @@
|
|||||||
import json
|
|
||||||
import logging
|
import logging
|
||||||
import urllib.request
|
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from typing import TYPE_CHECKING, Any
|
from typing import TYPE_CHECKING, Any
|
||||||
|
|
||||||
|
import requests
|
||||||
|
|
||||||
from strix.config import load_settings
|
from strix.config import load_settings
|
||||||
from strix.telemetry._common import (
|
from strix.telemetry._common import (
|
||||||
SESSION_ID,
|
SESSION_ID,
|
||||||
@@ -37,13 +37,7 @@ def _send(event: str, properties: dict[str, Any]) -> bool:
|
|||||||
"distinct_id": SESSION_ID,
|
"distinct_id": SESSION_ID,
|
||||||
"properties": properties,
|
"properties": properties,
|
||||||
}
|
}
|
||||||
req = urllib.request.Request( # noqa: S310
|
requests.post(f"{_POSTHOG_HOST}/capture/", json=payload, timeout=10)
|
||||||
f"{_POSTHOG_HOST}/capture/",
|
|
||||||
data=json.dumps(payload).encode(),
|
|
||||||
headers={"Content-Type": "application/json"},
|
|
||||||
)
|
|
||||||
with urllib.request.urlopen(req, timeout=10): # noqa: S310 # nosec B310
|
|
||||||
pass
|
|
||||||
except Exception: # noqa: BLE001
|
except Exception: # noqa: BLE001
|
||||||
logger.debug("posthog send failed for event %s", event, exc_info=True)
|
logger.debug("posthog send failed for event %s", event, exc_info=True)
|
||||||
return False
|
return False
|
||||||
|
|||||||
@@ -2,10 +2,11 @@ from __future__ import annotations
|
|||||||
|
|
||||||
import logging
|
import logging
|
||||||
import urllib.parse
|
import urllib.parse
|
||||||
import urllib.request
|
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from typing import TYPE_CHECKING, Any
|
from typing import TYPE_CHECKING, Any
|
||||||
|
|
||||||
|
import requests
|
||||||
|
|
||||||
from strix.config import load_settings
|
from strix.config import load_settings
|
||||||
from strix.telemetry._common import (
|
from strix.telemetry._common import (
|
||||||
SESSION_ID,
|
SESSION_ID,
|
||||||
@@ -42,9 +43,7 @@ def _send(event: str, properties: dict[str, Any]) -> bool:
|
|||||||
url = f"{_SCARF_ENDPOINT}{path}"
|
url = f"{_SCARF_ENDPOINT}{path}"
|
||||||
if query:
|
if query:
|
||||||
url = f"{url}?{query}"
|
url = f"{url}?{query}"
|
||||||
req = urllib.request.Request(url, method="POST") # noqa: S310
|
requests.post(url, timeout=10)
|
||||||
with urllib.request.urlopen(req, timeout=10): # noqa: S310 # nosec B310
|
|
||||||
pass
|
|
||||||
except Exception: # noqa: BLE001
|
except Exception: # noqa: BLE001
|
||||||
logger.debug("scarf send failed for event %s", event, exc_info=True)
|
logger.debug("scarf send failed for event %s", event, exc_info=True)
|
||||||
return False
|
return False
|
||||||
|
|||||||
@@ -7,8 +7,10 @@ import hashlib
|
|||||||
import json
|
import json
|
||||||
import time
|
import time
|
||||||
from typing import TYPE_CHECKING, Any
|
from typing import TYPE_CHECKING, Any
|
||||||
|
from unittest import mock
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
import requests
|
||||||
|
|
||||||
from strix.config import codex
|
from strix.config import codex
|
||||||
|
|
||||||
@@ -52,6 +54,18 @@ def test_authorize_url_carries_pkce_and_client() -> None:
|
|||||||
assert "state=st8" in url
|
assert "state=st8" in url
|
||||||
|
|
||||||
|
|
||||||
|
def test_post_form_returns_parsed_body() -> None:
|
||||||
|
resp = mock.MagicMock()
|
||||||
|
resp.status_code = 200
|
||||||
|
resp.content = b'{"access_token": "tok"}'
|
||||||
|
|
||||||
|
with mock.patch.object(requests, "post", return_value=resp) as post:
|
||||||
|
data = codex._post_form({"grant_type": "refresh_token"})
|
||||||
|
|
||||||
|
assert data == {"access_token": "tok"}
|
||||||
|
assert post.call_args.kwargs["timeout"] == codex._TOKEN_TIMEOUT
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
("value", "expected"),
|
("value", "expected"),
|
||||||
[
|
[
|
||||||
|
|||||||
@@ -2411,7 +2411,7 @@ wheels = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "strix-agent"
|
name = "strix-agent"
|
||||||
version = "1.4.0"
|
version = "1.4.1"
|
||||||
source = { editable = "." }
|
source = { editable = "." }
|
||||||
dependencies = [
|
dependencies = [
|
||||||
{ name = "caido-sdk-client" },
|
{ name = "caido-sdk-client" },
|
||||||
|
|||||||
Reference in New Issue
Block a user