152 lines
7.7 KiB
Python
152 lines
7.7 KiB
Python
import io
|
|
import json
|
|
import unittest
|
|
import urllib.error
|
|
from contextlib import redirect_stderr, redirect_stdout
|
|
|
|
from tests.test_gitea import FakeOpener, FakeResponse
|
|
from werkrevisor import secretprobe
|
|
|
|
VALUE = "probe-secret-0123456789abcdef"
|
|
|
|
|
|
def answer(body):
|
|
return lambda request: FakeResponse(json.dumps(body).encode())
|
|
|
|
|
|
def status(code):
|
|
def raise_it(request):
|
|
raise urllib.error.HTTPError(request.full_url, code, "error", {}, None)
|
|
return raise_it
|
|
|
|
|
|
def probe(key, *handlers):
|
|
queue = list(handlers)
|
|
opener = FakeOpener(lambda request: queue.pop(0)(request))
|
|
out = io.StringIO()
|
|
with redirect_stdout(out):
|
|
code = secretprobe.main([key], io.StringIO(VALUE + "\n"), opener)
|
|
assert VALUE not in out.getvalue()
|
|
return code, out.getvalue(), opener.requests
|
|
|
|
|
|
USER = answer({"login": "ai-reviewer"})
|
|
REPO = answer({"permissions": {"admin": False, "push": True}})
|
|
|
|
|
|
class GiteaProbeTest(unittest.TestCase):
|
|
def test_the_bot_with_write_access_and_write_scopes(self):
|
|
code, out, requests = probe("GITEA_BOT_TOKEN", USER, REPO, status(404), status(422))
|
|
self.assertEqual((code, out), (0, "GITEA_BOT_TOKEN: accepted: ai-reviewer, with write access to mi/werkprobe "
|
|
"and the scopes to write repository and issue\n"))
|
|
self.assertEqual([(r.get_method(), r.full_url) for r in requests], [
|
|
("GET", "https://git.javagil.de/api/v1/user"),
|
|
("GET", "https://git.javagil.de/api/v1/repos/mi/werkprobe"),
|
|
("POST", "https://git.javagil.de/api/v1/repos/mi/werkprobe/pulls/0/reviews"),
|
|
("POST", "https://git.javagil.de/api/v1/repos/mi/werkprobe/issues/0/comments")])
|
|
self.assertTrue(all(r.get_header("Authorization") == f"token {VALUE}" for r in requests))
|
|
|
|
def test_a_token_with_read_scopes_only(self):
|
|
self.assertEqual(probe("GITEA_BOT_TOKEN", USER, REPO, status(403))[:2],
|
|
(1, "GITEA_BOT_TOKEN: ai-reviewer, but the token may not write repository (HTTP 403)\n"))
|
|
self.assertEqual(probe("GITEA_BOT_TOKEN", USER, REPO, status(404), status(403))[:2],
|
|
(1, "GITEA_BOT_TOKEN: ai-reviewer, but the token may not write issue (HTTP 403)\n"))
|
|
|
|
def test_a_write_that_succeeds_is_no_probe(self):
|
|
self.assertEqual(probe("GITEA_BOT_TOKEN", USER, REPO, answer({"id": 1}))[:2],
|
|
(2, "GITEA_BOT_TOKEN: HTTP 200 on writing repository, the probe did not complete\n"))
|
|
|
|
def test_a_token_of_another_account(self):
|
|
code, out, _ = probe("GITEA_BOT_TOKEN", answer({"login": "mi"}))
|
|
self.assertEqual((code, out), (1, "GITEA_BOT_TOKEN: the token belongs to 'mi', not to ai-reviewer\n"))
|
|
|
|
def test_an_account_without_write_access(self):
|
|
code, out, _ = probe("GITEA_BOT_TOKEN", USER, answer({"permissions": {"push": False}}))
|
|
self.assertEqual((code, out), (1, "GITEA_BOT_TOKEN: ai-reviewer, but without write access to mi/werkprobe\n"))
|
|
|
|
def test_a_revoked_token(self):
|
|
self.assertEqual(probe("GITEA_BOT_TOKEN", status(401))[:2], (1, "GITEA_BOT_TOKEN: refused with HTTP 401\n"))
|
|
|
|
def test_an_outage_is_no_refusal(self):
|
|
self.assertEqual(probe("GITEA_BOT_TOKEN", status(502))[:2],
|
|
(2, "GITEA_BOT_TOKEN: HTTP 502, the probe did not complete\n"))
|
|
self.assertEqual(probe("GITEA_BOT_TOKEN", USER, status(503))[:2],
|
|
(2, "GITEA_BOT_TOKEN: HTTP 503, the probe did not complete\n"))
|
|
|
|
def unreachable(request):
|
|
raise urllib.error.URLError("Name or service not known")
|
|
self.assertEqual(probe("GITEA_BOT_TOKEN", unreachable)[:2],
|
|
(2, "GITEA_BOT_TOKEN: GET /api/v1/user failed: Name or service not known\n"))
|
|
|
|
|
|
class OpenRouterProbeTest(unittest.TestCase):
|
|
def test_the_key_info_without_its_label(self):
|
|
code, out, requests = probe("OPENROUTER_API_KEY", answer({"data": {
|
|
"label": "sk-or-v1-abc...xyz", "limit": 100, "limit_remaining": 63.214, "limit_reset": "monthly"}}))
|
|
self.assertEqual((code, out), (0, "OPENROUTER_API_KEY: accepted: limit 100.00 USD, 63.21 USD left, "
|
|
"limit reset monthly\n"))
|
|
self.assertEqual(requests[0].get_header("Authorization"), f"Bearer {VALUE}")
|
|
|
|
def test_a_key_without_limit(self):
|
|
self.assertEqual(probe("OPENROUTER_API_KEY", answer({"data": {"limit": None, "limit_remaining": None}}))[1],
|
|
"OPENROUTER_API_KEY: accepted: limit none, none left, limit reset never\n")
|
|
|
|
def test_a_revoked_key(self):
|
|
self.assertEqual(probe("OPENROUTER_API_KEY", status(401))[:2], (1, "OPENROUTER_API_KEY: refused with HTTP 401\n"))
|
|
|
|
|
|
class TypeSafeProbeTest(unittest.TestCase):
|
|
def test_the_model_list_names_jev(self):
|
|
code, out, requests = probe("TYPESAFE_API_KEY", answer({"models": [{"name": "jev-latest"}, {"name": "jev-preview"}]}))
|
|
self.assertEqual((code, out), (0, "TYPESAFE_API_KEY: accepted: the account has jev-latest\n"))
|
|
self.assertEqual((requests[0].full_url, requests[0].get_method()), ("https://api.typesafe.ai/v1/models", "GET"))
|
|
self.assertEqual(requests[0].get_header("Authorization"), f"Bearer {VALUE}")
|
|
|
|
def test_an_account_without_jev(self):
|
|
self.assertEqual(probe("TYPESAFE_API_KEY", answer({"models": []}))[:2],
|
|
(1, "TYPESAFE_API_KEY: accepted, but the account lacks jev-latest\n"))
|
|
|
|
def test_a_revoked_key(self):
|
|
self.assertEqual(probe("TYPESAFE_API_KEY", status(401))[:2], (1, "TYPESAFE_API_KEY: refused with HTTP 401\n"))
|
|
|
|
|
|
class StatsProbeTest(unittest.TestCase):
|
|
def test_a_source_token_gets_the_empty_record_refused(self):
|
|
code, out, requests = probe("WERKREVISOR_STATS_TOKEN", status(400))
|
|
self.assertEqual((code, out), (0, "WERKREVISOR_STATS_TOKEN: accepted as source token; "
|
|
"the empty record was refused, nothing stored\n"))
|
|
self.assertEqual((requests[0].full_url, requests[0].data), ("https://werkrevisor.javagil.de/api/usage", b"{}"))
|
|
|
|
def test_an_unknown_token(self):
|
|
self.assertEqual(probe("WERKREVISOR_STATS_TOKEN", status(401))[:2],
|
|
(1, "WERKREVISOR_STATS_TOKEN: refused: no source token of the service\n"))
|
|
|
|
def test_a_stored_record_would_be_no_probe(self):
|
|
self.assertEqual(probe("WERKREVISOR_STATS_TOKEN", answer({"stored": True}))[0], 2)
|
|
|
|
|
|
class MalformedValueTest(unittest.TestCase):
|
|
def test_a_value_with_a_line_break_never_reaches_a_header_or_the_output(self):
|
|
# The real opener: http.client rejects the header before it connects, with the value in its message.
|
|
for value in ("first-line-0123456789\nsecond-line", "with space 0123456789", "x1y2z3"):
|
|
out, err = io.StringIO(), io.StringIO()
|
|
with redirect_stdout(out), redirect_stderr(err):
|
|
code = secretprobe.main(["GITEA_BOT_TOKEN"], io.StringIO(value + "\n"))
|
|
self.assertEqual((code, out.getvalue()), (2, ""), value)
|
|
self.assertEqual(err.getvalue(), "GITEA_BOT_TOKEN: the value is too short, too long or has characters "
|
|
"other than A-Z a-z 0-9 . _ ~ + / = -\n")
|
|
for part in value.split():
|
|
self.assertNotIn(part, err.getvalue())
|
|
|
|
|
|
class UsageTest(unittest.TestCase):
|
|
def test_unknown_key_or_empty_value(self):
|
|
with redirect_stderr(io.StringIO()) as err:
|
|
self.assertEqual(secretprobe.main(["GITEA_CLAUDE_CODE_TOKEN"], io.StringIO(VALUE)), 2)
|
|
self.assertEqual(secretprobe.main(["GITEA_BOT_TOKEN"], io.StringIO("\n")), 2)
|
|
self.assertNotIn(VALUE, err.getvalue())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|