fix(cli): the login needs an encrypted connection, and a grade is an integer 1..5

Two CodeRabbit findings on the credential-routing change.

- _origin() accepted http://, so a trusted-but-cleartext AMP would still
  have received the bearer token in a header. The credential now also
  requires an encrypted connection: HTTPS, or plain HTTP to this machine
  (localhost, its subdomains, loopback), which is the rule
  TraceGrantClient already applies to collector grants. Anything else
  reads the run anonymously and says which of the two reasons applies.
- A verdict's grades were checked with isinstance(grade, int), which
  accepts True and 6; both would have printed as real grades and let the
  command exit 0. A grade is now an exact int in 1..5, or null.

Tests: a trusted http origin gets no credential, localhost does, the
encryption rule itself over nine origins, and four more malformed verdicts
(bool, 6, 0, 4.5). 56 passed.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Joao Moura
2026-09-21 08:34:42 -07:00
parent 0f6e499df5
commit 0185dc7a22
2 changed files with 95 additions and 9 deletions

View File

@@ -15,6 +15,7 @@ traces. The command sends the saved `crewai login` when there is one.
from __future__ import annotations
import contextlib
from ipaddress import ip_address
import json
import os
from pathlib import Path
@@ -99,19 +100,46 @@ def _origin(url: str | None) -> str | None:
)
def _encrypted(origin: str | None) -> bool:
"""HTTPS, or plain HTTP to this machine — the rule `TraceGrantClient` already
applies to collector grants (localhost, its subdomains, loopback addresses)."""
parsed = urlparse(origin or "")
if parsed.scheme == "https":
return True
if parsed.scheme != "http":
return False
hostname = (parsed.hostname or "").rstrip(".")
if hostname == "localhost" or hostname.endswith(".localhost"):
return True
try:
return ip_address(hostname).is_loopback
except ValueError:
return False
def _amp_client(trusted: set[str]) -> PlusAPI:
"""The AMP to ask, and whether the saved login goes with it.
A project's `.env` may point `crewai eval` at another AMP — that is how a
self-hosted project is wired, and the run was traced there — so the request
follows it. The credential does not: it goes only to an AMP this machine is
logged in to, and the run is read anonymously anywhere else."""
logged in to, over a connection that encrypts it. Anywhere else the run is
read anonymously."""
client = PlusAPI(api_key=saved_login())
if client.api_key is None or _origin(client.base_url) in trusted:
if client.api_key is None:
return client
origin = _origin(client.base_url)
if origin in trusted and _encrypted(origin):
return client
why = (
"is not an AMP this machine is logged in to"
if origin not in trusted
else "would carry the login over plain HTTP"
)
console.print(
f"Reading anonymously: {client.base_url} is not an AMP this machine is logged in to. "
f"Reading anonymously: {client.base_url} {why}. "
"Run `crewai enterprise configure <url>` to log in to it.",
style="yellow",
)
@@ -271,18 +299,21 @@ def _wait(client: PlusAPI, evaluation_id: str, url: str | None) -> dict[str, Any
def _well_formed_verdict(verdict: Any) -> bool:
"""`{"gate": "<word>", "grades": {area: 1..5 | null}}` — anything else is a protocol error."""
"""`{"gate": "<word>", "grades": {area: 1..5 | null}}` — anything else is a
protocol error. A grade is an exact integer in range: `True` is an `int` to
Python and 6 is not a grade, and neither may print as one."""
return (
isinstance(verdict, dict)
and isinstance(verdict.get("gate"), str)
and isinstance(verdict.get("grades"), dict)
and all(
grade is None or isinstance(grade, int)
for grade in verdict["grades"].values()
)
and all(_a_grade(grade) for grade in verdict["grades"].values())
)
def _a_grade(grade: Any) -> bool:
return grade is None or (type(grade) is int and 1 <= grade <= 5)
def _print_verdict(finished: dict[str, Any], url: str | None) -> None:
if finished.get("status") != "done":
console.print(

View File

@@ -187,6 +187,49 @@ def test_a_project_may_point_at_another_amp_but_never_gets_the_saved_login(proje
assert "crewai enterprise configure" in out
def test_a_trusted_amp_over_plain_http_still_gets_no_credential(project, monkeypatch, capsys):
"""A cleartext connection is not a place to put a bearer token, trusted or not."""
directory, _ = project
monkeypatch.setenv("CREWAI_PLUS_URL", "http://amp.internal") # exported, so it IS trusted
record_last_run(directory, amp_base_url="http://amp.internal")
amp = install(monkeypatch, FakeAMP(statuses=[done()]))
eval_module.eval_crew()
assert amp.base_url == "http://amp.internal" and amp.api_key is None
assert "would carry the login over plain HTTP" in capsys.readouterr().out
def test_plain_http_to_this_machine_is_fine_for_local_development(project, monkeypatch, capsys):
directory, _ = project
monkeypatch.setenv("CREWAI_PLUS_URL", "http://localhost:3000")
record_last_run(directory, amp_base_url="http://localhost:3000")
amp = install(monkeypatch, FakeAMP(statuses=[done()]))
eval_module.eval_crew()
assert amp.api_key == "login-token"
assert "Reading anonymously" not in capsys.readouterr().out
@pytest.mark.parametrize(
("origin", "encrypted"),
[
("https://app.crewai.com", True),
("http://localhost:3000", True),
("http://127.0.0.1:8000", True),
("http://[::1]:8000", True),
("http://amp.localhost", True),
("http://amp.internal", False),
("http://169.254.169.254", False),
("ftp://amp.test", False),
(None, False),
],
)
def test_which_connections_may_carry_the_login(origin, encrypted):
assert eval_module._encrypted(origin) is encrypted
def test_an_amp_exported_in_this_shell_is_trusted(project, monkeypatch, capsys):
directory, _ = project
monkeypatch.setenv("CREWAI_PLUS_URL", "https://shell.amp.test") # exported before the project is read
@@ -216,7 +259,19 @@ def test_an_origin_is_scheme_and_host_only(url, origin):
@pytest.mark.parametrize(
"verdict",
[None, "passed", [], {"gate": "passed"}, {"gate": None, "grades": {}}, {"gate": "passed", "grades": "5/5"}, {"gate": "passed", "grades": {"goal": "five"}}],
[
None,
"passed",
[],
{"gate": "passed"},
{"gate": None, "grades": {}},
{"gate": "passed", "grades": "5/5"},
{"gate": "passed", "grades": {"goal": "five"}},
{"gate": "passed", "grades": {"goal": True}}, # a bool is an int to Python, never a grade
{"gate": "passed", "grades": {"goal": 6}}, # out of the 1..5 range
{"gate": "passed", "grades": {"goal": 0}},
{"gate": "passed", "grades": {"goal": 4.5}},
],
)
def test_a_done_answer_without_a_well_formed_verdict_is_a_protocol_error(project, monkeypatch, capsys, verdict):
directory, _ = project