Skip to content
Merged
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
12 changes: 12 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,18 @@ All notable changes to the AxonFlow Python SDK will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).

## [6.2.0] - 2026-04-09

### Added

- **Telemetry `endpoint_type` field.** The anonymous telemetry ping now includes an SDK-derived classification of the configured AxonFlow endpoint as one of `localhost`, `private_network`, `remote`, or `unknown`. The raw URL is never sent and is not hashed. This helps distinguish self-hosted evaluation from real production deployments on the checkpoint dashboard. Opt out as before via `DO_NOT_TRACK=1` or `AXONFLOW_TELEMETRY=off`.

### Changed

- Examples and documentation updated to reflect the new AxonFlow platform v6.2.0 defaults for `PII_ACTION` (now `warn` — was `redact`) and the new `AXONFLOW_PROFILE` env var. No SDK API changes; the SDK continues to pass `PII_ACTION` through unchanged.

---

## [6.1.0] - 2026-04-06

### Added
Expand Down
2 changes: 1 addition & 1 deletion axonflow/_version.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
"""Single source of truth for the AxonFlow SDK version."""

__version__ = "6.1.0"
__version__ = "6.2.0"
60 changes: 58 additions & 2 deletions axonflow/telemetry.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,11 +13,13 @@

from __future__ import annotations

import ipaddress
import logging
import os
import platform
import threading
import uuid
from urllib.parse import urlparse

import httpx

Expand Down Expand Up @@ -74,6 +76,54 @@ def _detect_platform_version(endpoint: str) -> str | None:
return None


# Loopback and any-interface addresses. "0.0.0.0" is intentionally included
# here because it's the canonical bind-all-interfaces address and, in the
# context of an AxonFlow client endpoint, means "talk to localhost".
# noqa: S104 is scoped to the tuple below — this is not a bind operation.
_LOCALHOST_HOSTS = frozenset({"localhost", "127.0.0.1", "::1", "0.0.0.0"}) # noqa: S104


def _classify_endpoint(url: str | None) -> str: # noqa: PLR0911
"""Classify the configured AxonFlow endpoint for analytics (#1525).

Returns one of:
``"localhost"`` — localhost, 127.0.0.1, ::1, 0.0.0.0, ``*.localhost``
``"private_network"`` — RFC1918 ranges, link-local, ``*.local``,
``*.internal``, ``*.lan``, ``*.intranet``
``"remote"`` — everything else
``"unknown"`` — on any parse failure

The raw URL is never sent — only the classification. See issue #1525.
"""
if not url:
return "unknown"
try:
host = urlparse(url).hostname
except (ValueError, AttributeError):
return "unknown"
if not host:
return "unknown"
host = host.lower()

if host in _LOCALHOST_HOSTS or host.endswith(".localhost"):
return "localhost"

if any(host.endswith(suffix) for suffix in (".local", ".internal", ".lan", ".intranet")):
return "private_network"

# Try parsing as an IP address (v4 or v6).
try:
ip = ipaddress.ip_address(host)
except ValueError:
# Not an IP; treat remaining hostnames as remote.
return "remote"
if ip.is_loopback:
return "localhost"
if ip.is_private or ip.is_link_local:
return "private_network"
return "remote"


def _normalize_arch(arch: str) -> str:
"""Normalize architecture names to match other SDKs."""
if arch == "aarch64":
Expand All @@ -83,7 +133,11 @@ def _normalize_arch(arch: str) -> str:
return arch


def _build_payload(mode: str, platform_version: str | None = None) -> dict[str, object]:
def _build_payload(
mode: str,
platform_version: str | None = None,
endpoint_type: str = "unknown",
) -> dict[str, object]:
"""Build the JSON payload for the checkpoint ping."""
return {
"sdk": "python",
Expand All @@ -93,6 +147,7 @@ def _build_payload(mode: str, platform_version: str | None = None) -> dict[str,
"arch": _normalize_arch(platform.machine()),
"runtime_version": platform.python_version(),
"deployment_mode": mode,
"endpoint_type": endpoint_type,
"features": [],
"instance_id": str(uuid.uuid4()),
}
Expand All @@ -102,7 +157,8 @@ def _do_ping(url: str, mode: str, endpoint: str, debug: bool) -> None:
"""Execute the HTTP POST (runs inside a daemon thread)."""
try:
platform_version = _detect_platform_version(endpoint) if endpoint else None
payload = _build_payload(mode, platform_version)
endpoint_type = _classify_endpoint(endpoint)
payload = _build_payload(mode, platform_version, endpoint_type)
resp = httpx.post(url, json=payload, timeout=_TIMEOUT_SECONDS)
if resp.status_code == _HTTP_OK:
try:
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

[project]
name = "axonflow"
version = "6.0.0"
version = "6.2.0"
description = "AxonFlow Python SDK - Enterprise AI Governance in 3 Lines of Code"
readme = "README.md"
license = {text = "MIT"}
Expand Down
70 changes: 70 additions & 0 deletions tests/test_telemetry_endpoint_type.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
"""Unit tests for _classify_endpoint (issue #1525)."""

from axonflow.telemetry import _build_payload, _classify_endpoint


class TestClassifyEndpoint:
def test_empty_string_is_unknown(self):
assert _classify_endpoint("") == "unknown"

def test_none_is_unknown(self):
assert _classify_endpoint(None) == "unknown"

def test_localhost_variants(self):
assert _classify_endpoint("http://localhost:8080") == "localhost"
assert _classify_endpoint("https://localhost") == "localhost"
assert _classify_endpoint("http://127.0.0.1") == "localhost"
assert _classify_endpoint("http://127.0.0.1:8080") == "localhost"
assert _classify_endpoint("http://[::1]") == "localhost"
assert _classify_endpoint("http://0.0.0.0:8080") == "localhost"
assert _classify_endpoint("http://agent.localhost") == "localhost"

def test_private_network_ipv4(self):
# RFC1918
assert _classify_endpoint("http://10.0.0.1") == "private_network"
assert _classify_endpoint("http://10.1.2.3") == "private_network"
assert _classify_endpoint("http://172.16.0.1") == "private_network"
assert _classify_endpoint("http://172.31.255.254") == "private_network"
assert _classify_endpoint("http://192.168.1.1") == "private_network"
# Link-local
assert _classify_endpoint("http://169.254.169.254") == "private_network"

def test_private_network_hostnames(self):
assert _classify_endpoint("http://agent.internal") == "private_network"
assert _classify_endpoint("http://agent.local") == "private_network"
assert _classify_endpoint("http://agent.lan") == "private_network"
assert _classify_endpoint("http://agent.intranet") == "private_network"

def test_remote(self):
assert _classify_endpoint("https://production-us.getaxonflow.com") == "remote"
assert _classify_endpoint("https://checkpoint.getaxonflow.com") == "remote"
assert _classify_endpoint("https://api.example.com") == "remote"
assert _classify_endpoint("http://8.8.8.8") == "remote"

def test_malformed_url(self):
# urlparse treats "not-a-url" as a scheme-less path, hostname=None → unknown.
assert _classify_endpoint("not-a-url") == "unknown"
assert _classify_endpoint("://nohost") == "unknown"

def test_case_insensitive(self):
assert _classify_endpoint("http://LOCALHOST:8080") == "localhost"
assert _classify_endpoint("http://AGENT.INTERNAL") == "private_network"


class TestPayloadIncludesEndpointType:
def test_payload_default_endpoint_type_is_unknown(self):
payload = _build_payload("production")
assert payload["endpoint_type"] == "unknown"

def test_payload_with_explicit_endpoint_type(self):
payload = _build_payload("production", None, "localhost")
assert payload["endpoint_type"] == "localhost"

def test_payload_does_not_contain_raw_url(self):
"""Critical non-goal: the SDK must never send the configured URL."""
payload = _build_payload("production", None, "remote")
payload_str = str(payload)
# Assert no URL-like strings in the payload.
assert "http://" not in payload_str
assert "https://" not in payload_str
assert "getaxonflow.com" not in payload_str
Loading