External/violin
1
0
Fork 0
mirror of https://github.com/Strategic-Automation/violin.git synced 2026-08-14 12:33:37 +02:00
Code Issues Packages Projects Releases Wiki Activity

Make burst admission atomic (#61)

Browse Source 
Closes #11
This commit is contained in:
Dan
2026-08-01 22:20:47 +01:00
committed by GitHub
parent e5e258fa0b
commit 7607bcb181
6 changed files with 192 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+1
View File
@@ -2,6 +2,7 @@
## 3.0.0
- Made burst execution atomic at admission: every command is preflighted before launch, required sync credit is reserved under one lock, and unused reservations are returned after partial batches.
- Fixed target extraction for dotted identifiers and direct `/dev/tcp`/`/dev/udp` redirections, and prevented network-capable local-looking commands from bypassing execution accounting.
- Made interrupted skill preparation recoverable with expiring reservations and stale-owner protection; batch review now remains tied to the delivered execution receipt.
- Stabilized the core engagement workflow: domain/URL-only scopes now validate, runtime execution cannot substitute another scope file, PTT/review CLI contracts carry skill metadata, and review reuses the active delivered binding.
plugins/violin_guard/execution.py
+42 -6
View File
@@ -248,10 +248,30 @@ def _monitor_background(
)
def _commit_started_command(engagement: Path, command: str, phase: str, ptt_task_id: str) -> int:
def _commit_started_command(
engagement: Path,
command: str,
phase: str,
ptt_task_id: str,
sync_reservation: str | None = None,
) -> tuple[int, bool]:
if state.is_local_bookkeeping_command(command):
return state.sync_credit_remaining(str(engagement), phase)
return _commit_guard_state(engagement, command, phase, ptt_task_id)
return state.sync_credit_remaining(str(engagement), phase), False
if sync_reservation:
state.record_ok_check(str(engagement), command, phase)
remaining = state.consume_reserved_sync_credit(str(engagement), sync_reservation)
state.mark_pending_sync(str(engagement), command, phase, ptt_task_id)
count = state.tick_command(str(engagement))
from .phases import suppresses_heartbeat
phase_enum = normalize_phase(phase)
if count % state.COMMAND_INTERVAL == 0 and not suppresses_heartbeat(phase_enum):
state.set_heartbeat_pending(
str(engagement),
f"Reached {count} executed target commands. Review engagement files for drift.",
)
return remaining, True
return _commit_guard_state(engagement, command, phase, ptt_task_id), False
def _start_background_monitor(
@@ -265,12 +285,15 @@ def _start_background_monitor(
command: str,
phase: str,
ptt_task_id: str,
sync_reservation: str | None,
timeout: int,
execution_id: str,
) -> dict[str, Any]:
state.atomic_json(manifest_path, record)
try:
remaining = _commit_started_command(engagement, command, phase, ptt_task_id)
remaining, consumed = _commit_started_command(
engagement, command, phase, ptt_task_id, sync_reservation
)
except Exception:
_terminate_process(proc)
raise
@@ -296,6 +319,7 @@ def _start_background_monitor(
"stderr_preview": "",
"sync_required": remaining <= 0,
"sync_credit_remaining": remaining,
"sync_reservation_consumed": consumed,
}
@@ -312,6 +336,7 @@ def execute(
ptt_task_id: str = "",
argv: list[str] | None = None,
background: bool = False,
sync_reservation: str | None = None,
) -> dict[str, Any]:
"""Execute one already-authorized command and persist its complete receipt."""
engagement = _resolve_engagement(eng_dir)
@@ -390,6 +415,7 @@ def execute(
command=command,
phase=phase,
ptt_task_id=ptt_task_id,
sync_reservation=sync_reservation,
timeout=timeout,
execution_id=execution_id,
)
@@ -442,15 +468,25 @@ def execute(
append_history(engagement, command, phase, exit_code, rel_manifest)
remaining = _commit_started_command(engagement, command, phase, ptt_task_id)
if sync_reservation and proc is None:
remaining = state.release_reserved_sync_credit(str(engagement), sync_reservation)
consumed = False
released = True
else:
remaining, consumed = _commit_started_command(
engagement, command, phase, ptt_task_id, sync_reservation
)
released = False
return {
**receipt,
"executed": True,
"executed": proc is not None,
"stdout_preview": _preview(stdout_path),
"stderr_preview": _preview(stderr_path),
"sync_required": remaining <= 0,
"sync_credit_remaining": remaining,
"sync_reservation_consumed": consumed,
"sync_reservation_released": released,
}
plugins/violin_guard/handlers/exec_handlers.py
+45 -5
View File
@@ -103,8 +103,8 @@ def handle_exec_burst(a, **kwargs):
active_task = ptt.find_active_task(ptt.parse_ptt(_eng_path(eng_dir) / "state" / "ptt.md"))
active_task_id = active_task.id if active_task else ""
results = []
executed = 0
preflight = []
required_slots = 0
for idx, cmd in enumerate(cmds):
cmd_args = {
"command": cmd,
@@ -120,9 +120,8 @@ def handle_exec_burst(a, **kwargs):
if status_name == "block":
return _json(
"denied",
executed=executed,
results=results
+ [
executed=0,
results=[
{
"index": idx + 1,
"command": cmd,
@@ -133,6 +132,31 @@ def handle_exec_burst(a, **kwargs):
reason=f"command [{idx + 1}] blocked: {r.errors[0] if r.errors else 'blocked'}",
)
review_warnings = r.warnings if status_name == "review" else []
local = state.is_local_bookkeeping_command(cmd)
if not local:
required_slots += 1
preflight.append(
{
"index": idx + 1,
"command": cmd,
"review_warnings": review_warnings,
"local": local,
}
)
reservation_id = None
if required_slots:
try:
reservation_id = state.reserve_sync_credit(eng_dir, phase, required_slots)
except ValueError as exc:
return _json("denied", executed=0, results=[], reason=str(exc))
results = []
executed = 0
for item in preflight:
idx = item["index"]
cmd = item["command"]
review_warnings = item["review_warnings"]
try:
res = execution.execute(
command=cmd,
@@ -143,6 +167,7 @@ def handle_exec_burst(a, **kwargs):
cwd=cwd,
label=label,
ptt_task_id=active_task_id,
sync_reservation=None if item["local"] else reservation_id,
)
res.pop("status", None)
entry = {"index": idx + 1, "command": cmd, **res}
@@ -152,9 +177,21 @@ def handle_exec_burst(a, **kwargs):
results.append(entry)
if res.get("executed"):
executed += 1
if (
reservation_id
and not item["local"]
and not res.get("sync_reservation_consumed")
and not res.get("sync_reservation_released")
):
if res.get("executed"):
state.consume_reserved_sync_credit(eng_dir, reservation_id)
else:
state.release_reserved_sync_credit(eng_dir, reservation_id)
if res.get("exit_code", 0) != 0 and not continue_on_error:
break
except Exception as e: # noqa: BLE001
if reservation_id:
state.release_reserved_sync_credit(eng_dir, reservation_id)
if not continue_on_error:
return _json(
"execution_failed",
@@ -164,6 +201,9 @@ def handle_exec_burst(a, **kwargs):
)
results.append({"index": idx + 1, "command": cmd, "error": str(e)})
if reservation_id:
state.release_reserved_sync_credit(eng_dir, reservation_id)
return _json(
"batch_complete",
executed=executed,
plugins/violin_guard/state.py
+54
View File
@@ -211,6 +211,59 @@ def spend_sync_credit(eng_dir: str | Path, phase: str) -> int:
return mutate_json(path, spend)
def reserve_sync_credit(eng_dir: str | Path, phase: str, count: int) -> str:
"""Atomically reserve credit for a burst before any command starts."""
if count < 1:
raise ValueError("a sync reservation must contain at least one command")
path = _sync_path(eng_dir)
def reserve(data: dict[str, Any]) -> str:
credit = max(0, int(data.get("credit", sync_credit_limit(phase))))
if credit < count:
raise ValueError(f"insufficient sync credit for burst: need {count}, have {credit}")
reservation_id = f"burst-{uuid.uuid4().hex}"
data["credit"] = credit - count
data.setdefault("reservations", {})[reservation_id] = {
"phase": phase,
"remaining": count,
"created_at": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
}
return reservation_id
return mutate_json(path, reserve)
def consume_reserved_sync_credit(eng_dir: str | Path, reservation_id: str) -> int:
"""Consume one previously reserved slot without decrementing credit twice."""
path = _sync_path(eng_dir)
def consume(data: dict[str, Any]) -> int:
reservation = (data.get("reservations") or {}).get(reservation_id)
if not reservation or int(reservation.get("remaining", 0)) < 1:
raise ValueError("sync reservation is missing or exhausted")
reservation["remaining"] = int(reservation["remaining"]) - 1
if reservation["remaining"] == 0:
data["reservations"].pop(reservation_id, None)
return max(0, int(data.get("credit", 0)))
return mutate_json(path, consume)
def release_reserved_sync_credit(eng_dir: str | Path, reservation_id: str) -> int:
"""Return every unconsumed slot in a reservation to the sync window."""
path = _sync_path(eng_dir)
def release(data: dict[str, Any]) -> int:
reservation = (data.get("reservations") or {}).pop(reservation_id, None)
if reservation:
data["credit"] = int(data.get("credit", 0)) + max(
0, int(reservation.get("remaining", 0))
)
return max(0, int(data.get("credit", 0)))
return mutate_json(path, release)
def mark_pending_sync(
eng_dir: str | Path,
command: str,
@@ -248,6 +301,7 @@ def clear_pending_sync(eng_dir: str | Path) -> None:
def clear(data: dict[str, Any]) -> None:
data.pop("pending", None)
data.pop("credit", None)
data.pop("reservations", None)
mutate_json(path, clear)
tests/guard/state/test_burst_and_target.py
+40 -5
View File
@@ -24,6 +24,7 @@ from plugins.violin_guard import (
handlers as service,
)
from plugins.violin_guard import handlers as tools # noqa: E402
from plugins.violin_guard.command import CheckResult # noqa: E402
from plugins.violin_guard.targets import resolve_target # noqa: E402
from tests.guard.receipt_fixture import bind_active_task # noqa: E402
@@ -186,9 +187,16 @@ def _patch_burst(monkeypatch, eng_dir):
def fake_execute(command, *, eng_dir=eng_dir, phase, **kwargs):
rec["commands"].append(command)
active = ptt.find_active_task(ptt.parse_ptt(Path(eng_dir) / "state" / "ptt.md"))
remaining = execution._commit_guard_state(
Path(eng_dir), command, phase, active.id if active else ""
)
reservation_id = kwargs.get("sync_reservation")
if reservation_id:
state.record_ok_check(eng_dir, command, phase)
remaining = state.consume_reserved_sync_credit(eng_dir, reservation_id)
state.mark_pending_sync(eng_dir, command, phase, active.id if active else "")
state.tick_command(eng_dir)
else:
remaining = execution._commit_guard_state(
Path(eng_dir), command, phase, active.id if active else ""
)
rec["batch_id"] = state.get_pending_sync(eng_dir)
return {
"execution_id": "00000000-0000-0000-0000-000000000001",
@@ -207,6 +215,7 @@ def _patch_burst(monkeypatch, eng_dir):
"evidence_paths": {},
"sync_required": remaining <= 0,
"sync_credit_remaining": remaining,
"sync_reservation_consumed": bool(reservation_id),
}
monkeypatch.setattr(execution, "execute", fake_execute)
@@ -288,8 +297,34 @@ def test_exec_burst_fail_closed_on_blocked_command(eng, monkeypatch):
assert (
data["reason"] == "command [2] blocked: destructive filesystem deletion (rm -rf) is blocked"
), data
# First command ran; the blocked one did not, and nothing after it ran.
assert rec["commands"] == ["nmap -sV 10.10.10.10"]
# Preflight is atomic: the blocked command prevents every command from launching.
assert rec["commands"] == []
def test_exec_burst_preflights_every_command_before_launch(eng, monkeypatch):
from plugins.violin_guard.handlers import exec_handlers
checks = iter((CheckResult(), CheckResult(errors=["blocked second command"])))
launched: list[str] = []
monkeypatch.setattr(exec_handlers, "_check_command_internal", lambda _args: next(checks))
monkeypatch.setattr(execution, "execute", lambda command, **_kwargs: launched.append(command))
before = state.sync_credit_remaining(eng, "recon")
result = json.loads(
service.handle_exec_burst(
{
"eng_dir": str(eng),
"phase": "recon",
"target": "10.10.10.10",
"commands": ["first", "second"],
}
)
)
assert result["status"] == "denied"
assert result["executed"] == 0
assert launched == []
assert state.sync_credit_remaining(eng, "recon") == before
def test_exec_burst_missing_commands_file(eng):
tests/guard/state/test_sync_credit_window.py
+10
View File
@@ -37,6 +37,16 @@ def test_network_clients_are_not_local_bookkeeping() -> None:
assert not state.is_local_bookkeeping_command("cat < /dev/tcp/10.10.10.10/80")
def test_sync_credit_reservation_consumes_and_releases_atomically(tmp_path: Path) -> None:
eng = _engagement(tmp_path)
before = state.sync_credit_remaining(eng, "recon")
reservation = state.reserve_sync_credit(eng, "recon", 2)
assert state.sync_credit_remaining(eng, "recon") == before - 2
state.consume_reserved_sync_credit(eng, reservation)
state.release_reserved_sync_credit(eng, reservation)
assert state.sync_credit_remaining(eng, "recon") == before - 1
def test_phase_window_runs_without_yolo_then_next_command_blocks(
monkeypatch, tmp_path: Path
) -> None: