Compare commits

..

6 Commits

Author SHA1 Message Date
didericis-claude 217eadf9a1 fix(dlp): skip projection passes when exact variant is safe-listed
lint / lint (push) Failing after 2m8s
test / unit (pull_request) Successful in 43s
test / integration (pull_request) Successful in 25s
When a supervisor-approved safe-token exactly matched an env secret
(Pass 1), Passes 2 & 3 (alnum projection) still ran and re-blocked on
the same value.  Track whether any variant was found-and-approved and
skip the projection passes for that secret in that case.
2026-06-24 22:45:51 -04:00
didericis-claude 3fe3829c8d docs(prd): flip prd-new-strengthen-outbound-exfil-detection Draft → Active
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-24 22:45:51 -04:00
didericis-claude 51751c8d28 feat(egress): inject per-session canary token into sidecar and agent environments
EgressPlan gains a `canary: str` field (default "") populated in Egress.prepare()
using secrets.token_urlsafe(32).  Each launched bottle:

  - sidecar receives EGRESS_TOKEN_CANARY=<value> (literal env entry, scanned by
    existing known-secrets detector without any detector code changes)
  - agent receives BOT_BOTTLE_CANARY=<value> (visible fake secret that signals
    exfiltration with zero false positives if it appears in outbound traffic)

Docker compose and macos-container backends updated; smolmachines shares docker
compose and so picks this up automatically.  Unit tests cover canary uniqueness,
detection via scan_known_secrets, and EgressPlan backward-compat default.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-24 22:45:51 -04:00
didericis-claude 330e836085 feat(dlp): fragmentation resistance, entropy detector, broadened known-value scan
- _alnum_projection(): strip non-alphanumeric chars for separator-injection detection
- scan_known_secrets() gains two extra passes per secret after exact-variant matching:
  alnum-projection exact match (catches hyphens/spaces between secret chars) and a
  sliding-window partial-match scan (catches chunked substrings ≥ PARTIAL_MATCH_MIN_LEN)
- scan_known_secrets() accepts sensitive_prefixes param (default ("EGRESS_TOKEN_",))
  so redact_tokens and call-sites can extend the scanned env-var prefix set
- scan_entropy() warn-only detector flagging windows with Shannon entropy ≥ 5.5 bits/char
- "entropy" added to OUTBOUND_DETECTOR_NAMES; scan_outbound opts it in only when
  explicitly listed in dlp.outbound_detectors (never part of the default "all" set)
- scan_outbound reads BOT_BOTTLE_SENSITIVE_PREFIXES from environ to extend
  scan_known_secrets beyond EGRESS_TOKEN_* without schema changes
- Binary bodies decoded via latin-1 fallback (bijective byte↔codepoint) instead
  of utf-8 errors=replace, preserving ASCII secret strings in binary payloads

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-24 22:45:51 -04:00
didericis-claude fa38012621 docs: draft PRD prd-new for strengthen-outbound-exfil-detection
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-24 22:45:51 -04:00
didericis-claude 2e790268b0 fix(deploy-key): raise DeployKeyCollisionError on 422 key conflicts
lint / lint (push) Successful in 2m7s
test / unit (push) Successful in 46s
test / integration (push) Successful in 25s
Update Quality Badges / update-badges (push) Successful in 2m7s
Gitea returns HTTP 422 when a deploy key title or public key content
already exists on the repo. The provisioner previously surfaced this
as a generic RuntimeError with the raw status code. Introduce
DeployKeyCollisionError (a RuntimeError subclass) in the base module
and detect 422 in GiteaDeployKeyProvisioner.create so callers can
catch collisions explicitly and the error message names the repo and
title involved.
2026-06-25 02:23:12 +00:00
3 changed files with 35 additions and 1 deletions
@@ -19,7 +19,7 @@ import urllib.error
import urllib.request
from pathlib import Path
from ...deploy_key_provisioner import DeployKeyProvisioner
from ...deploy_key_provisioner import DeployKeyCollisionError, DeployKeyProvisioner
class GiteaDeployKeyProvisioner(DeployKeyProvisioner):
@@ -71,6 +71,11 @@ class GiteaDeployKeyProvisioner(DeployKeyProvisioner):
body = json.loads(resp.read())
except urllib.error.HTTPError as exc:
_body = _read_error_body(exc)
if exc.code == 422:
raise DeployKeyCollisionError(
f"deploy key collision for {owner_repo!r} "
f"(title={title!r}): key title or content already registered — {_body}"
) from exc
raise RuntimeError(
f"failed to create deploy key for {owner_repo}: "
f"HTTP {exc.code}{_body}"
+4
View File
@@ -11,6 +11,10 @@ from __future__ import annotations
from abc import ABC, abstractmethod
class DeployKeyCollisionError(RuntimeError):
"""Raised when a deploy key title or public key already exists on the repo."""
class DeployKeyProvisioner(ABC):
"""Manages a single deploy-key lifecycle on a remote forge."""
@@ -12,6 +12,7 @@ from bot_bottle.contrib.gitea.deploy_key_provisioner import (
GiteaDeployKeyProvisioner,
_split_owner_repo,
)
from bot_bottle.deploy_key_provisioner import DeployKeyCollisionError
def _provisioner() -> GiteaDeployKeyProvisioner:
@@ -100,6 +101,30 @@ class TestCreate(unittest.TestCase):
provisioner.create("owner/repo", "title")
self.assertIn("403", str(ctx.exception))
def test_create_raises_collision_error_on_422(self):
provisioner = _provisioner()
collision_body = json.dumps({
"errors": ["Key content already exists on this repository"],
"message": "422 Unprocessable Entity",
})
with patch(
"bot_bottle.contrib.gitea.deploy_key_provisioner.subprocess.run"
), patch(
"bot_bottle.contrib.gitea.deploy_key_provisioner.urllib.request.urlopen",
side_effect=_http_error(422, collision_body),
), patch(
"bot_bottle.contrib.gitea.deploy_key_provisioner.Path.read_bytes",
return_value=b"pk",
), patch(
"bot_bottle.contrib.gitea.deploy_key_provisioner.Path.read_text",
return_value="ssh-ed25519 AAAA\n",
):
with self.assertRaises(DeployKeyCollisionError) as ctx:
provisioner.create("owner/repo", "my-title")
msg = str(ctx.exception)
self.assertIn("owner/repo", msg)
self.assertIn("my-title", msg)
class TestDelete(unittest.TestCase):
def test_delete_calls_correct_endpoint(self):