Files
bot-bottle/tests/unit/test_yaml_subset.py
T
didericis 70f773ac61
test / unit (pull_request) Successful in 17s
test / integration (pull_request) Successful in 1m3s
feat(egress-proxy): cutover from cred-proxy (PRD 0017 chunk 2)
Hard cutover. cred-proxy is deleted; egress-proxy is now the agent's
HTTP_PROXY (when routes are declared) with pipelock on its outbound
leg. Two per-bottle CAs are minted: egress-proxy's (agent trust
store) and pipelock's (egress-proxy's outbound trust store).

Manifest:
  - `bottle.cred_proxy` → hard error with a migration recipe.
  - `bottle.egress_proxy` is the new shape (PRD 0017 chunk 1).
  - CredProxy* types + role validators removed.

Wiring:
  - launch.py: `egress_proxy_tls_init` mints the egress-proxy CA
    (cert+key concat for mitmproxy + cert-only for agent trust);
    `DockerEgressProxy.start` docker-cps both CAs in, sets
    `HTTPS_PROXY=pipelock` + `EGRESS_PROXY_UPSTREAM_CA` so mitmdump
    trusts pipelock's MITM. Agent's HTTP_PROXY points at
    egress-proxy when routes exist, else falls back to pipelock
    (no-routes bottles unchanged).
  - prepare.py / backend.py: `cred_proxy` arg → `egress_proxy`;
    sidecar-orphan probe + plan field + dashboard view all
    renamed.
  - provision_ca: selects the egress-proxy CA when present, else
    pipelock's (filename renamed to claude-bottle-mitm-ca.crt).
  - bottle.provision: cred-proxy dotfile rewrites (~/.npmrc,
    ~/.gitconfig insteadOf, tea config) are gone — HTTP_PROXY
    catches everything respecting it.

Pipelock helpers:
  - `pipelock_token_hosts` → `pipelock_route_hosts` (now reading
    egress_proxy.routes).
  - cred-proxy hostname auto-allow → egress-proxy hostname
    auto-allow.
  - Anthropic seed-phrase workaround now triggers when an
    egress_proxy route targets api.anthropic.com (was based on the
    cred-proxy `anthropic-base-url` role).

Dockerfile.egress-proxy:
  - Entrypoint conditionally passes
    `--set ssl_verify_upstream_trusted_ca=$EGRESS_PROXY_UPSTREAM_CA`
    (via the `${VAR:+...}` shell expansion) so standalone runs without
    a mounted pipelock CA still boot.
  - mkdirs `/home/mitmproxy/.mitmproxy` ahead of `docker cp`.

Deleted: claude_bottle/{cred_proxy,cred_proxy_server}.py,
backend/docker/{cred_proxy,provision/cred_proxy}.py,
Dockerfile.cred-proxy, plus the corresponding unit + integration
tests. backend/docker/cred_proxy_apply.py stays as a stub for
chunk 3 to rewrite (its container-name + routes-path constants
are inlined so it survives without the deleted module).

Test changes:
  - test_pipelock_allowlist rewritten against egress-proxy routes
    + the new `pipelock_route_hosts`.
  - test_manifest_md_load + test_pipelock_yaml + test_yaml_subset
    fixtures migrated to the `egress_proxy: { routes: [...] }`
    shape.
  - test_supervise_sidecar's round-trip test switched from
    `dashboard.approve` to `dashboard.reject`: the approval-apply
    path on cred-proxy-block proposals hits a deleted sidecar in
    chunk 2's transitional state. Chunk 3 restores the approval
    test once the remediation flow is retargeted at egress-proxy.

376 tests pass (was 427; net delta is removed cred-proxy tests).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-25 14:30:39 -04:00

332 lines
9.2 KiB
Python

"""Unit: YAML-subset parser used by the per-file MD manifest
(PRD 0011). Covers happy paths, the constructs the manifest files
actually use, and every rejection case the PRD enumerates."""
import textwrap
import unittest
from claude_bottle.log import Die
from claude_bottle.yaml_subset import parse_frontmatter, parse_yaml_subset
def _y(s: str):
"""Parse a dedented YAML string."""
return parse_yaml_subset(textwrap.dedent(s).lstrip("\n"))
class TestScalars(unittest.TestCase):
def test_string(self):
self.assertEqual({"k": "hello"}, _y("k: hello\n"))
def test_string_with_url_chars(self):
self.assertEqual(
{"k": "https://example.com/path?x=1"},
_y("k: https://example.com/path?x=1\n"),
)
def test_int(self):
self.assertEqual({"port": 9099}, _y("port: 9099\n"))
def test_negative_int(self):
self.assertEqual({"n": -3}, _y("n: -3\n"))
def test_bool_true(self):
self.assertEqual({"x": True}, _y("x: true\n"))
def test_bool_false(self):
self.assertEqual({"x": False}, _y("x: false\n"))
def test_null(self):
self.assertEqual({"x": None}, _y("x: null\n"))
def test_tilde_null(self):
self.assertEqual({"x": None}, _y("x: ~\n"))
def test_double_quoted_string(self):
self.assertEqual({"k": "a b"}, _y('k: "a b"\n'))
def test_double_quoted_escape(self):
self.assertEqual({"k": "a\nb"}, _y(r'k: "a\nb"' + "\n"))
def test_single_quoted_string(self):
self.assertEqual({"k": "a b"}, _y("k: 'a b'\n"))
def test_single_quoted_apos_double(self):
# Single-quoted YAML uses `''` to embed a literal `'`.
self.assertEqual({"k": "it's"}, _y("k: 'it''s'\n"))
class TestForbiddenBoolLikes(unittest.TestCase):
"""Ambiguous bool-ish tokens have to be quoted explicitly."""
def _expect_die(self, src: str):
with self.assertRaises(Die):
_y(src)
def test_yes_dies(self):
self._expect_die("k: yes\n")
def test_no_dies(self):
self._expect_die("k: no\n")
def test_on_dies(self):
self._expect_die("k: on\n")
def test_capital_TRUE_dies(self):
self._expect_die("k: TRUE\n")
def test_norway_quoted_is_fine(self):
self.assertEqual({"country": "NO"}, _y('country: "NO"\n'))
class TestForbiddenScalarShapes(unittest.TestCase):
def _expect_die(self, src: str):
with self.assertRaises(Die):
_y(src)
def test_bare_date_dies(self):
self._expect_die("k: 2026-05-24\n")
def test_bare_octal_dies(self):
self._expect_die("k: 0o755\n")
def test_bare_hex_dies(self):
self._expect_die("k: 0xFF\n")
def test_bare_float_dies(self):
self._expect_die("k: 1.5\n")
def test_quoted_date_is_fine(self):
self.assertEqual({"k": "2026-05-24"}, _y('k: "2026-05-24"\n'))
class TestMapping(unittest.TestCase):
def test_flat_mapping(self):
self.assertEqual(
{"a": 1, "b": "two", "c": True},
_y("""
a: 1
b: two
c: true
"""),
)
def test_nested_mapping(self):
out = _y("""
outer:
inner: hello
other: 5
""")
self.assertEqual({"outer": {"inner": "hello", "other": 5}}, out)
def test_duplicate_key_dies(self):
with self.assertRaises(Die):
_y("""
a: 1
a: 2
""")
def test_key_must_be_bare_identifier(self):
with self.assertRaises(Die):
_y('"weird key": 1\n')
class TestBlockList(unittest.TestCase):
def test_list_of_strings(self):
out = _y("""
allowlist:
- example.com
- github.com
""")
self.assertEqual({"allowlist": ["example.com", "github.com"]}, out)
def test_list_of_mappings(self):
out = _y("""
routes:
- path: /a/
upstream: https://a.example
- path: /b/
upstream: https://b.example
""")
self.assertEqual(
{"routes": [
{"path": "/a/", "upstream": "https://a.example"},
{"path": "/b/", "upstream": "https://b.example"},
]},
out,
)
def test_list_item_with_nested_mapping(self):
out = _y("""
entries:
- name: foo
ExtraHosts:
host.example: 10.0.0.1
- name: bar
""")
self.assertEqual(
{"entries": [
{"name": "foo", "ExtraHosts": {"host.example": "10.0.0.1"}},
{"name": "bar"},
]},
out,
)
def test_list_item_with_inline_list_value(self):
# role: [git-insteadof, tea-login] — the exact shape in the
# claude-bottle manifest.
out = _y("""
routes:
- path: /x/
role: [git-insteadof, tea-login]
""")
self.assertEqual(
{"routes": [
{"path": "/x/", "role": ["git-insteadof", "tea-login"]},
]},
out,
)
class TestInline(unittest.TestCase):
def test_inline_list(self):
self.assertEqual({"l": [1, 2, 3]}, _y("l: [1, 2, 3]\n"))
def test_inline_list_of_strings(self):
self.assertEqual({"l": ["a", "b", "c"]}, _y("l: [a, b, c]\n"))
def test_inline_dict(self):
self.assertEqual(
{"d": {"a": "1", "b": "2"}},
_y('d: {a: "1", b: "2"}\n'),
)
def test_nested_flow_dies(self):
with self.assertRaises(Die):
_y("l: [[1, 2], [3, 4]]\n")
class TestForbiddenConstructs(unittest.TestCase):
def test_anchor_dies(self):
with self.assertRaises(Die):
_y("""
a: &anchor 1
b: *anchor
""")
def test_multiline_block_scalar_dies(self):
with self.assertRaises(Die):
_y("""
k: |
line 1
line 2
""")
def test_tag_dies(self):
with self.assertRaises(Die):
_y("k: !!str hello\n")
def test_tab_in_indent_dies(self):
with self.assertRaises(Die):
_y("a:\n\tb: 1\n")
class TestComments(unittest.TestCase):
def test_full_line_comment(self):
out = _y("""
# comment
k: v
""")
self.assertEqual({"k": "v"}, out)
def test_trailing_comment(self):
self.assertEqual({"k": "v"}, _y("k: v # trailing\n"))
def test_hash_in_quoted_string_kept(self):
self.assertEqual({"k": "a#b"}, _y('k: "a#b"\n'))
class TestRealisticBottleFile(unittest.TestCase):
"""The exact shape a real bottle frontmatter takes — the parser
has to round-trip this without surprise."""
def test_dev_bottle(self):
out = _y("""
egress_proxy:
routes:
- host: api.anthropic.com
auth:
scheme: Bearer
token_ref: CLAUDE_CODE_OAUTH_TOKEN
- host: gitea.dideric.is
auth:
scheme: token
token_ref: GITEA_TOKEN
path_allowlist:
- /didericis/
git:
- Name: claude-bottle
Upstream: ssh://git@gitea.dideric.is:30009/x/y.git
IdentityFile: ~/.ssh/gitea.pem
ExtraHosts:
gitea.dideric.is: 100.78.141.42
egress:
allowlist:
- example.com
""")
# Spot-check the deep parts; the structure is large.
self.assertEqual(2, len(out["egress_proxy"]["routes"]))
self.assertEqual(
["/didericis/"],
out["egress_proxy"]["routes"][1]["path_allowlist"],
)
self.assertEqual(
"Bearer",
out["egress_proxy"]["routes"][0]["auth"]["scheme"],
)
self.assertEqual(
"100.78.141.42",
out["git"][0]["ExtraHosts"]["gitea.dideric.is"],
)
self.assertEqual(["example.com"], out["egress"]["allowlist"])
class TestFrontmatter(unittest.TestCase):
def test_basic(self):
text = textwrap.dedent("""
---
bottle: dev
---
This is the body.
""").lstrip("\n")
fm, body = parse_frontmatter(text)
self.assertEqual({"bottle": "dev"}, fm)
self.assertIn("This is the body", body)
def test_no_frontmatter_passes_through(self):
text = "no frontmatter here\njust body\n"
fm, body = parse_frontmatter(text)
self.assertEqual({}, fm)
self.assertEqual(text, body)
def test_unclosed_frontmatter_dies(self):
with self.assertRaises(Die):
parse_frontmatter("---\nbottle: dev\nno closing")
def test_body_preserves_blank_lines(self):
text = (
"---\n"
"k: v\n"
"---\n"
"\n"
"line one\n"
"\n"
"line three\n"
)
_, body = parse_frontmatter(text)
self.assertEqual("\nline one\n\nline three\n", body)
if __name__ == "__main__":
unittest.main()