rob
/
CmdForge
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Harden M9: Ed25519 attestation, full SHA-256 integrity, locked-by-default, prompt improvements, registry verification

This commit is contained in:
rob 2026-07-20 16:50:49 -03:00
parent 04153a4bbd
commit 56cb34276e
34 changed files with 2549 additions and 257 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

69
README.md
View File

@ -123,6 +123,7 @@ Opens the graphical interface where you can create and manage tools visually. Fe
- **My Tools** - Browse, create, edit, and delete tools organized by category
- **Registry** - Search and install community tools from the CmdForge registry
- **Providers** - Manage AI provider configurations
- **Reuse & Discovery** - Compare similar tools and preview safe extraction of reusable steps
### CLI Mode
@ -178,10 +179,26 @@ cmdforge settings mytool diff # Show changes from defaults
cmdforge config show # Show current config
cmdforge config connect username # Connect to registry account
cmdforge config disconnect # Disconnect from registry
# Private, local pipeline suggestions (disabled by default)
cmdforge usage enable # Begin recording tool names in shell pipes
cmdforge usage suggestions # Show frequent pipelines
cmdforge usage clear # Delete all local usage history
# Model Context Protocol (optional: pip install -e ".[mcp]")
cmdforge mcp add local --command npx --arg=-y --arg @scope/server
cmdforge mcp add remote --transport streamable-http --url https://example.com/mcp
cmdforge mcp connect remote
cmdforge mcp serve # Expose approved tools over stdio
cmdforge mcp serve --transport streamable-http # Loopback-only HTTP by default
```
`cf` searches the public registry when no local tool matches. Registry results
show available relevance and quality evidence and install on selection.
When local pipeline discovery is explicitly enabled, `cf` also suggests
frequently repeated pipe pairs. Only tool names, anonymous pipe identifiers,
counts, and timestamps are kept in `~/.cmdforge/usage.json`; command arguments
and input/output content are never recorded or transmitted.
### Running Tools
@ -664,6 +681,9 @@ cmdforge registry search --owner official --min-downloads 100
# Connect your account (opens browser for authentication)
cmdforge config connect yourusername
# Create an Ed25519 release key and register its public half
cmdforge registry signing-key init
# Publish a tool
cmdforge registry publish mytool
@ -678,6 +698,51 @@ cmdforge registry status --sync
```
Published tools go through moderation before appearing publicly. You'll receive feedback if changes are requested.
Once a signing key is registered, publishing signs the transitive content
identity automatically. The registry rejects unsigned or invalid releases and
clients verify the signature before installing. The private key remains in
`~/.cmdforge/release-signing-key.json` with `0600` permissions.
### Prompt Optimization
Generate local deterministic prompt variations without contacting a provider:
```bash
cmdforge optimize summarize -n 5
```
Structural contract checks cannot rank prompt meaning, so CmdForge will not
claim a winning variation unless explicit behavioral cases are supplied:
```json
[
{"input": "A long input", "contains": "summary"},
{"input": "Another input", "expected": {"result": "expected value"}}
]
```
```bash
cmdforge optimize summarize --behavior-tests behavior.json \
--test-provider opencode-pickle
```
Behavior tests execute every candidate and may incur provider cost. Use
`--provider NAME` separately when you explicitly want a provider to generate a
rephrased candidate.
### Community Improvements
```bash
cmdforge registry improve owner/tool 1.0.0 0 proposed-prompt.txt \
--rationale "Clearer output requirement"
cmdforge registry review-improvement 42 approve --notes "Tests pass"
# Apply the approved text, bump the tool version, then publish it:
cmdforge registry publish mytool --improvement-id 42
```
Submissions are auto-tested before review. Contributor credit and the
`optimized`/`community-reviewed` badges are awarded only when a published
version actually contains the approved change.
### Tool Documentation
@ -750,6 +815,10 @@ The graphical interface provides a modern desktop experience:
### Tool Builder
- Visual form for creating and editing tools
- Reuse & Discovery panel with asynchronous registry similarity results,
exact local reuse evidence, and opt-in pipeline suggestions
- Extraction creates a separate tool only after showing a unified diff; the
current draft is never rewritten automatically
- Add arguments with flags and default values
- Add prompt steps (AI calls) with profile selection
- Add code steps with **AI-assisted code generation**:

1
pyproject.toml
View File

@ -36,6 +36,7 @@ dependencies = [
"NodeGraphQt>=0.6.0",
"setuptools", # Required for distutils compatibility (Python 3.12+)
"jsonschema>=4.0", # JSON schema validation for structured output
"cryptography>=41.0", # Ed25519 release attestations
]
[project.optional-dependencies]

177
src/cmdforge/attestation.py
View File

@ -1,28 +1,52 @@
"""Supply chain attestation (M9.5).
"""Ed25519 supply-chain attestations for CmdForge releases."""
Tool publishers sign releases. Registry verifies signatures before
accepting publish. Clients verify signatures before installing.
"""
from __future__ import annotations
import hashlib
import hmac
import base64
import json
from dataclasses import dataclass, field
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Dict, Optional
from typing import Dict, Optional, Tuple
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
Ed25519PrivateKey,
Ed25519PublicKey,
)
@dataclass
def _canonical_payload(
tool_name: str,
version: str,
content_hash: str,
signer: str,
signed_at: str,
algorithm: str,
) -> bytes:
return json.dumps(
{
"algorithm": algorithm,
"content_hash": content_hash,
"signed_at": signed_at,
"signer": signer,
"tool_name": tool_name,
"version": version,
},
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
@dataclass(frozen=True)
class Attestation:
"""A signed attestation for a tool release."""
tool_name: str
version: str
content_hash: str
signer: str # publisher identity
signer: str
signature: str
signed_at: str = ""
algorithm: str = "hmac-sha256"
signed_at: str
algorithm: str = "ed25519"
def to_dict(self) -> dict:
return {
@ -35,76 +59,89 @@ class Attestation:
"algorithm": self.algorithm,
}
@classmethod
def from_dict(cls, value: dict) -> "Attestation":
return cls(**{key: value[key] for key in (
"tool_name", "version", "content_hash", "signer", "signature",
"signed_at", "algorithm",
)})
def generate_keypair() -> Tuple[str, str]:
"""Generate base64-encoded raw Ed25519 private/public keys."""
private = Ed25519PrivateKey.generate()
private_bytes = private.private_bytes(
serialization.Encoding.Raw,
serialization.PrivateFormat.Raw,
serialization.NoEncryption(),
)
public_bytes = private.public_key().public_bytes(
serialization.Encoding.Raw,
serialization.PublicFormat.Raw,
)
return (
base64.b64encode(private_bytes).decode("ascii"),
base64.b64encode(public_bytes).decode("ascii"),
)
def sign_tool(
tool_name: str,
version: str,
content_hash: str,
signer: str,
secret_key: str,
private_key: str,
*,
signed_at: Optional[str] = None,
) -> Attestation:
"""Sign a tool release with HMAC-SHA256.
Args:
tool_name: Tool name
version: Tool version
content_hash: Content hash from integrity module
signer: Publisher identity (username)
secret_key: Signing key (from config or keyring)
Returns:
Attestation with signature
"""
payload = f"{tool_name}:{version}:{content_hash}:{signer}"
signature = hmac.new(
secret_key.encode(),
payload.encode(),
hashlib.sha256,
).hexdigest()
"""Sign release identity using a publisher's private Ed25519 key."""
timestamp = signed_at or datetime.now(timezone.utc).isoformat()
algorithm = "ed25519"
payload = _canonical_payload(
tool_name, version, content_hash, signer, timestamp, algorithm
)
key = Ed25519PrivateKey.from_private_bytes(base64.b64decode(private_key))
signature = base64.b64encode(key.sign(payload)).decode("ascii")
return Attestation(
tool_name=tool_name,
version=version,
content_hash=content_hash,
signer=signer,
signature=signature,
signed_at=datetime.now(timezone.utc).isoformat(),
tool_name, version, content_hash, signer, signature, timestamp, algorithm
)
def verify_attestation(attestation: Attestation, secret_key: str) -> bool:
"""Verify a tool attestation signature.
def verify_attestation(attestation: Attestation, public_key: str) -> bool:
"""Verify every signed field using the publisher's trusted public key."""
if attestation.algorithm != "ed25519":
return False
try:
key = Ed25519PublicKey.from_public_bytes(base64.b64decode(public_key))
signature = base64.b64decode(attestation.signature, validate=True)
payload = _canonical_payload(
attestation.tool_name,
attestation.version,
attestation.content_hash,
attestation.signer,
attestation.signed_at,
attestation.algorithm,
)
key.verify(signature, payload)
return True
except (ValueError, TypeError, InvalidSignature):
return False
Args:
attestation: The attestation to verify
secret_key: The signing key to verify against
Returns:
True if signature is valid
"""
payload = f"{attestation.tool_name}:{attestation.version}:{attestation.content_hash}:{attestation.signer}"
expected = hmac.new(
secret_key.encode(),
payload.encode(),
hashlib.sha256,
).hexdigest()
return hmac.compare_digest(expected, attestation.signature)
def verify_trusted_attestation(
attestation: Attestation, trusted_publishers: Dict[str, str]
) -> bool:
"""Anchor an attestation to an explicitly trusted publisher key."""
public_key = trusted_publishers.get(attestation.signer)
return bool(public_key) and verify_attestation(attestation, public_key)
def verify_content_hash(content_hash: str, tool_dict: dict) -> bool:
"""Verify that a content hash matches the tool definition.
Args:
content_hash: The hash to verify
tool_dict: Tool dictionary (from to_dict())
Returns:
True if hash matches
"""
"""Verify a definition hash against a serialized Tool definition."""
from .integrity import compute_tool_hash
# Reconstruct what the hash should be
tool_dict_copy = dict(tool_dict)
tool_dict_copy.pop("path", None)
content = json.dumps(tool_dict_copy, sort_keys=True)
expected = hashlib.sha256(content.encode()).hexdigest()[:16]
return expected == content_hash
from .tool import Tool
try:
return compute_tool_hash(Tool.from_dict(tool_dict)) == content_hash
except (KeyError, TypeError, ValueError):
return False

104
src/cmdforge/cli/__init__.py
View File

@ -17,6 +17,7 @@ from .config_commands import cmd_config
from .settings_commands import cmd_settings
from .system_deps_commands import cmd_system_deps
from .mcp_commands import cmd_mcp
from .usage_commands import cmd_usage
def main():
@ -107,6 +108,17 @@ def main():
p_optimize = subparsers.add_parser("optimize", help="Generate and test prompt variations for a tool")
p_optimize.add_argument("name", help="Tool name")
p_optimize.add_argument("-n", "--count", type=int, default=5, help="Number of variations")
p_optimize.add_argument(
"--provider", default="mock",
help="Provider used to generate a variation (default: local deterministic mock)",
)
p_optimize.add_argument(
"--behavior-tests", metavar="FILE",
help="JSON cases used to score semantic behavior (executes the tool)",
)
p_optimize.add_argument(
"--test-provider", help="Provider override used while running behavior tests",
)
p_optimize.set_defaults(func=cmd_optimize)
# 'check' command
@ -229,8 +241,40 @@ def main():
p_reg_publish.add_argument("--dry-run", action="store_true", help="Validate without publishing")
p_reg_publish.add_argument("-f", "--force", action="store_true", help="Skip confirmation prompts")
p_reg_publish.add_argument("--owner", default="", help="Owner override (admin only, e.g. 'official')")
p_reg_publish.add_argument(
"--improvement-id", type=int,
help="Credit an approved improvement contained in this version",
)
p_reg_publish.set_defaults(func=cmd_registry)
p_reg_signing = registry_sub.add_parser(
"signing-key", help="Initialize or inspect the release signing key"
)
p_reg_signing.add_argument(
"action", nargs="?", choices=["init", "status"], default="status"
)
p_reg_signing.set_defaults(func=cmd_registry)
p_reg_improve = registry_sub.add_parser(
"improve", help="Submit an auto-tested prompt/code improvement"
)
p_reg_improve.add_argument("tool", help="Tool as owner/name")
p_reg_improve.add_argument("version", help="Exact tool version")
p_reg_improve.add_argument("step_index", type=int, help="Step to replace")
p_reg_improve.add_argument("file", help="File containing proposed prompt/code")
p_reg_improve.add_argument("--rationale", default="")
p_reg_improve.set_defaults(func=cmd_registry)
p_reg_review_improvement = registry_sub.add_parser(
"review-improvement", help="Review a tested community improvement"
)
p_reg_review_improvement.add_argument("id", type=int)
p_reg_review_improvement.add_argument(
"decision", choices=["approve", "reject", "request_changes"]
)
p_reg_review_improvement.add_argument("--notes", default="")
p_reg_review_improvement.set_defaults(func=cmd_registry)
# registry update-readme
p_reg_update_readme = registry_sub.add_parser("update-readme", help="Update README for a published tool")
p_reg_update_readme.add_argument("tool", nargs="?", default="", help="Tool name (local name, will resolve owner)")
@ -477,8 +521,21 @@ def main():
# mcp serve
p_mcp_serve = mcp_sub.add_parser("serve", help="Start CmdForge as an MCP server")
p_mcp_serve.add_argument(
"--transport", choices=["stdio"], default="stdio",
help="Transport (M7.3 supports stdio)",
"--transport", choices=["stdio", "streamable-http"], default="stdio",
help="MCP server transport",
)
p_mcp_serve.add_argument("--host", default="127.0.0.1")
p_mcp_serve.add_argument("--port", type=int, default=8000)
p_mcp_serve.add_argument(
"--allowed-origin", action="append", default=[],
help="Allowed browser Origin; repeat as needed",
)
p_mcp_serve.add_argument(
"--auth-token", help="Bearer token (or ${ENV_NAME})",
)
p_mcp_serve.add_argument(
"--external-url",
help="Public HTTPS URL when binding beyond localhost (TLS proxy required)",
)
p_mcp_serve.set_defaults(func=cmd_mcp)
@ -495,10 +552,15 @@ def main():
p_mcp_add = mcp_sub.add_parser("add", help="Add an MCP server")
p_mcp_add.add_argument("name", help="Server name")
p_mcp_add.add_argument(
"--transport", choices=["stdio"], default="stdio",
help="Transport type (M7.2 supports stdio)",
"--transport", choices=["stdio", "streamable-http"], default="stdio",
help="Transport type",
)
p_mcp_add.add_argument("--command", help="Executable for stdio servers")
p_mcp_add.add_argument("--url", help="Endpoint for Streamable HTTP servers")
p_mcp_add.add_argument(
"--header", action="append", default=[], metavar="NAME=VALUE",
help="HTTP header; repeat as needed (supports ${NAME} references)",
)
p_mcp_add.add_argument(
"--arg", dest="server_args", action="append", default=[], metavar="VALUE",
help="Command argument; repeat for each argument (use --arg=-y for leading dashes)",
@ -525,6 +587,20 @@ def main():
# Default for mcp with no subcommand (list)
p_mcp.set_defaults(func=cmd_mcp, mcp_cmd="list")
p_usage = subparsers.add_parser(
"usage", help="Manage opt-in local pipeline discovery"
)
usage_sub = p_usage.add_subparsers(dest="usage_cmd")
for action in ("enable", "disable", "status", "clear"):
command = usage_sub.add_parser(action)
command.set_defaults(func=cmd_usage)
p_usage_suggestions = usage_sub.add_parser(
"suggestions", help="Show frequently composed tool pipelines"
)
p_usage_suggestions.add_argument("--threshold", type=int, default=3)
p_usage_suggestions.set_defaults(func=cmd_usage)
p_usage.set_defaults(func=cmd_usage, usage_cmd="status")
args = parser.parse_args()
# If no command, launch UI
@ -682,7 +758,7 @@ def cmd_inspect(args):
def cmd_optimize(args):
"""Generate and test prompt variations for a tool."""
from ..prompt_optimizer import optimize_tool
from ..prompt_optimizer import load_behavioral_evaluator, optimize_tool
from ..tool import load_tool
tool = load_tool(args.name)
@ -693,7 +769,21 @@ def cmd_optimize(args):
print(f"Optimizing '{tool.name}' — generating {args.count} prompt variations...")
print()
result = optimize_tool(tool, count=args.count)
if args.count < 1:
print("Error: --count must be at least 1.", file=sys.stderr)
return 2
evaluator = None
if args.behavior_tests:
try:
evaluator = load_behavioral_evaluator(
args.behavior_tests, args.test_provider
)
except (OSError, ValueError) as exc:
print(f"Error loading behavior tests: {exc}", file=sys.stderr)
return 2
result = optimize_tool(
tool, count=args.count, provider=args.provider, evaluator=evaluator
)
print(f"Baseline score: {result.baseline_score}")
print(f"Variations tested: {len(result.variations)}")
@ -708,7 +798,7 @@ def cmd_optimize(args):
print()
print("To apply this variation, edit the tool's config.yaml and update the prompt step.")
else:
print("No improvements found. The original prompt may already be optimal.")
print(result.note or "No improvements found.")
return 0

28
src/cmdforge/cli/mcp_commands.py
View File

@ -51,6 +51,8 @@ def _cmd_mcp_list(args):
print(f" Command: {_sanitize(command)}")
if s.cwd:
print(f" Working dir: {s.cwd}")
else:
print(f" URL: {_sanitize(s.url or '')}")
print(f" Approved: {'yes' if s.approved else 'no'}")
print(f" Timeout: {s.timeout}s")
print()
@ -98,6 +100,7 @@ def _cmd_mcp_add(args):
try:
server_env = _parse_env_entries(getattr(args, "env", []))
headers = _parse_header_entries(getattr(args, "header", []))
except ValueError as exc:
print(f"Error: {exc}")
return 1
@ -111,6 +114,8 @@ def _cmd_mcp_add(args):
name=name,
transport=getattr(args, "transport", "stdio"),
command=getattr(args, "command", None),
url=getattr(args, "url", None),
headers=headers,
args=getattr(args, "server_args", []) or [],
cwd=getattr(args, "cwd", None),
env=server_env,
@ -171,13 +176,34 @@ def _parse_env_entries(entries):
return result
def _parse_header_entries(entries):
result = {}
for entry in entries:
if "=" not in entry:
raise ValueError(f"invalid --header value '{entry}'; expected NAME=VALUE")
name, value = entry.split("=", 1)
if not name.strip() or any(char in name for char in "\r\n"):
raise ValueError(f"invalid HTTP header name '{name}'")
if any(char in value for char in "\r\n"):
raise ValueError(f"invalid HTTP header value for '{name}'")
result[name.strip()] = value
return result
def _cmd_mcp_serve(args):
"""Start CmdForge as an MCP server."""
from ..mcp_server import serve as start_server
transport = getattr(args, "transport", "stdio")
try:
start_server(transport=transport)
start_server(
transport=transport,
host=getattr(args, "host", "127.0.0.1"),
port=getattr(args, "port", 8000),
allowed_origins=getattr(args, "allowed_origin", []),
auth_token=getattr(args, "auth_token", None),
external_url=getattr(args, "external_url", None),
)
except ImportError as exc:
print(f"Error: {exc}")
return 1

13
src/cmdforge/cli/picker.py
View File

@ -479,6 +479,19 @@ def main():
# Use stderr for UI if stdout is piped, so tool output stays clean
_ui_out = sys.stderr if not sys.stdout.isatty() else sys.stdout
try:
from ..usage import get_suggestions
suggestions = get_suggestions()[:3]
if suggestions:
_write(f"{GREEN}Frequent pipelines you could save as tools:{RESET}\n")
for item in suggestions:
_write(
f" {' | '.join(item['tools'])} "
f"{DIM}({item['count']} uses){RESET}\n"
)
except (OSError, ValueError):
pass
try:
with TTYInput() as tty_input:
result = run_picker(tty_input)

89
src/cmdforge/cli/registry_commands.py
View File

@ -29,6 +29,12 @@ def cmd_registry(args):
return _cmd_registry_update(args)
elif args.registry_cmd == "publish":
return _cmd_registry_publish(args)
elif args.registry_cmd == "signing-key":
return _cmd_registry_signing_key(args)
elif args.registry_cmd == "improve":
return _cmd_registry_improve(args)
elif args.registry_cmd == "review-improvement":
return _cmd_registry_review_improvement(args)
elif args.registry_cmd == "update-readme":
return _cmd_registry_update_readme(args)
elif args.registry_cmd == "my-tools":
@ -51,6 +57,8 @@ def cmd_registry(args):
print(" info <tool> Show tool information")
print(" update Update local index cache")
print(" publish [path] Publish a tool")
print(" signing-key Manage Ed25519 release signing")
print(" improve Submit a community improvement")
print(" update-readme Update README for published tool(s)")
print(" my-tools List your published tools")
print(" status <tool> Check moderation status of a tool")
@ -598,6 +606,7 @@ def _cmd_registry_publish(args):
from ..tool import load_tool, Tool, ToolStep
dep_result = None
my_owner = ""
try:
client = get_client()
@ -737,6 +746,7 @@ def _cmd_registry_publish(args):
_print_quality_summary(preflight_result.get("quality"))
if remote_report.get("errors"):
return 1
release_content_hash = preflight_result.get("content_hash", "")
if sys.stdin.isatty() and not getattr(args, "force", False):
try:
if input("Publish this validated version? [y/N] ").strip().lower() != "y":
@ -751,7 +761,26 @@ def _cmd_registry_publish(args):
try:
client = get_client()
owner = getattr(args, "owner", "")
result = client.publish_tool(config_yaml, readme, defaults, owner=owner)
attestation = None
from ..signing import load_signing_key
signing_key = load_signing_key()
if signing_key:
if not release_content_hash:
print(
"Registry did not return a content identity; refusing to sign.",
file=sys.stderr,
)
return 1
from ..attestation import sign_tool
signer = owner or my_owner
attestation = sign_tool(
name, version, release_content_hash, signer, signing_key[0]
).to_dict()
result = client.publish_tool(
config_yaml, readme, defaults, owner=owner,
attestation=attestation,
improvement_id=getattr(args, "improvement_id", None),
)
pr_url = result.get("pr_url", "")
status = result.get("status", "")
@ -826,6 +855,64 @@ def _cmd_registry_publish(args):
return 0
def _cmd_registry_signing_key(args):
"""Initialize a local key and register its public half with the registry."""
from ..signing import SIGNING_KEY_FILE, initialize_signing_key, load_signing_key
from ..registry_client import RegistryError, get_client
if args.action == "status":
key = load_signing_key()
print(
f"Signing key: {SIGNING_KEY_FILE}"
if key else "No release signing key configured."
)
return 0 if key else 1
try:
_, public_key = initialize_signing_key()
get_client().set_signing_public_key(public_key)
except (OSError, ValueError, RegistryError) as exc:
print(f"Could not initialize signing key: {exc}", file=sys.stderr)
return 1
print(f"Release signing key initialized: {SIGNING_KEY_FILE}")
print("The private key is stored locally with mode 0600; back it up securely.")
return 0
def _cmd_registry_improve(args):
from ..registry_client import RegistryError, get_client
if "/" not in args.tool:
print("Tool must be specified as owner/name.", file=sys.stderr)
return 2
owner, name = args.tool.split("/", 1)
try:
proposed = Path(args.file).read_text(encoding="utf-8")
result = get_client().submit_improvement(
owner, name, args.version, args.step_index, proposed,
args.rationale,
)
except (OSError, RegistryError) as exc:
print(f"Could not submit improvement: {exc}", file=sys.stderr)
return 1
print(f"Improvement {result.get('id')} tested and submitted for review.")
return 0
def _cmd_registry_review_improvement(args):
from ..registry_client import RegistryError, get_client
try:
result = get_client().review_improvement(args.id, args.decision, args.notes)
except RegistryError as exc:
print(f"Could not review improvement: {exc}", file=sys.stderr)
return 1
print(f"Improvement {args.id}: {result.get('status')}")
if result.get("ready_to_apply"):
print(
"Publish the updated version with "
f"--improvement-id {args.id} to apply credit and badges."
)
return 0
def _cmd_registry_update_readme(args):
"""Update README for published tool(s) on the registry."""
from ..registry_client import RegistryError, get_client

29
src/cmdforge/cli/usage_commands.py Normal file
View File

@ -0,0 +1,29 @@
"""CLI for opt-in local usage discovery."""
from ..usage import clear_usage, get_suggestions, is_enabled, set_enabled
def cmd_usage(args):
action = getattr(args, "usage_cmd", None) or "status"
if action == "enable":
set_enabled(True)
print("Local pipeline discovery enabled. No input, output, arguments, or telemetry is recorded.")
return 0
if action == "disable":
set_enabled(False)
print("Local pipeline discovery disabled.")
return 0
if action == "clear":
clear_usage()
print("Local usage history cleared.")
return 0
if action == "suggestions":
suggestions = get_suggestions(getattr(args, "threshold", 3))
if not suggestions:
print("No frequent pipeline suggestions yet.")
return 0
for index, item in enumerate(suggestions, start=1):
print(f"{index}. {' | '.join(item['tools'])} ({item['count']} uses)")
return 0
print(f"Local pipeline discovery: {'enabled' if is_enabled() else 'disabled'}")
return 0

39
src/cmdforge/community.py
View File

@ -98,6 +98,9 @@ def run_submission_tests(submission: ImprovementSubmission, tool: Tool) -> dict:
from .contract_testing import run_contract_tests
if submission.tool_name != tool.name or submission.tool_version != (tool.version or ""):
raise ValueError("Submission targets a different tool or version")
# Baseline
baseline = run_contract_tests(tool)
baseline_passed = sum(
@ -110,9 +113,27 @@ def run_submission_tests(submission: ImprovementSubmission, tool: Tool) -> dict:
return {"error": "Step index out of range", "baseline_passed": baseline_passed}
step = tool_copy.steps[submission.step_index]
current = step.prompt if isinstance(step, PromptStep) else (
step.code if isinstance(step, CodeStep) else None
)
if current != submission.original:
return {
"error": "Original step has changed since submission",
"baseline_passed": baseline_passed,
}
if isinstance(step, PromptStep) and submission.step_type == "prompt":
step.prompt = submission.proposed
elif isinstance(step, CodeStep) and submission.step_type == "code":
try:
compile(submission.proposed, f"<{tool.name}:step-{submission.step_index}>", "exec")
except SyntaxError as exc:
result = {
"error": f"Proposed code has invalid syntax: {exc.msg}",
"baseline_passed": baseline_passed,
"passed_for_review": False,
}
submission.test_result = result
return result
step.code = submission.proposed
else:
return {"error": "Step type mismatch", "baseline_passed": baseline_passed}
@ -128,6 +149,18 @@ def run_submission_tests(submission: ImprovementSubmission, tool: Tool) -> dict:
"proposed_passed": proposed_passed,
"improvement": proposed_passed - baseline_passed,
"regressed": proposed_passed < baseline_passed,
"baseline_outcome": baseline.outcome,
"proposed_outcome": proposed_result.outcome,
"passed_for_review": (
(
proposed_result.outcome == "passed"
and proposed_passed >= baseline_passed
)
or (
submission.step_type == "code"
and proposed_result.outcome == "unsupported"
)
),
"details": [r.to_dict() for r in proposed_result.results],
}
@ -146,10 +179,12 @@ def review_submission(
if decision not in ("approve", "reject", "request_changes"):
raise ValueError("Decision must be: approve, reject, or request_changes")
if submission.status == "pending":
if submission.status != "tested" or not (
submission.test_result or {}
).get("passed_for_review"):
raise ValueError("Submission must be tested before review")
submission.status = decision
submission.status = "approved" if decision == "approve" else decision
return SubmissionReview(
submission=submission,
decision=decision,

294
src/cmdforge/gui/pages/tool_builder_page.py
View File

@ -1,5 +1,9 @@
"""Tool builder page - create and edit tools."""
import copy
import difflib
import re
import yaml
from PySide6.QtWidgets import (
@ -7,9 +11,9 @@ from PySide6.QtWidgets import (
QLineEdit, QTextEdit, QPlainTextEdit, QComboBox, QPushButton,
QGroupBox, QListWidget, QListWidgetItem, QLabel,
QMessageBox, QSplitter, QFrame, QStackedWidget,
QButtonGroup
QButtonGroup, QInputDialog
)
from PySide6.QtCore import Qt
from PySide6.QtCore import Qt, QThread, Signal, QTimer
from ...tool import (
Tool, ToolArgument, PromptStep, CodeStep, ToolStep,
@ -19,6 +23,84 @@ from ...tool import (
from ..widgets.icons import get_prompt_icon, get_code_icon, get_tool_icon
class SimilarToolsWorker(QThread):
"""Search the registry without blocking the tool builder."""
results_ready = Signal(str, list)
def __init__(self, query: str, parent=None):
super().__init__(parent)
self.query = query
def run(self):
try:
from ...registry_client import RegistryClient, RegistryError
try:
result = RegistryClient().search_tools(self.query, per_page=5)
except RegistryError:
result = None
self.results_ready.emit(self.query, list(result.data) if result else [])
except Exception:
# Guidance is advisory; an unavailable registry must never prevent
# local authoring.
self.results_ready.emit(self.query, [])
def build_extracted_tool(tool: Tool, opportunity: dict, name: str) -> Tool:
"""Build, but do not save, a reusable tool from an evidenced sequence."""
if opportunity.get("type") != "repeated_sequence":
raise ValueError("Only an internal repeated sequence can be extracted")
locations = opportunity.get("locations") or []
length = int(opportunity.get("length", 0))
if not locations or length < 2:
raise ValueError("Reuse evidence does not identify a valid sequence")
start = int(locations[0]) - 1
selected = tool.steps[start:start + length]
if len(selected) != length:
raise ValueError("Reuse evidence falls outside the current draft")
available = {"input", "settings"} | {
argument.variable for argument in tool.arguments
}
for step in selected:
if not isinstance(step, PromptStep):
raise ValueError("Only prompt-step sequences can be extracted safely")
references = set(re.findall(r"(?<!\{)\{([A-Za-z_]\w*)\}(?!\})", step.prompt or ""))
missing = references - available
if missing:
raise ValueError(
"The sequence depends on values produced outside its boundary: "
+ ", ".join(sorted(missing))
)
if step.output_var:
available.update(
value.strip() for value in step.output_var.split(",") if value.strip()
)
output_var = getattr(selected[-1], "output_var", "result") or "result"
output_schema = getattr(selected[-1], "output_schema", None)
return Tool(
name=name,
description=f"Extracted reusable sequence from {tool.name}",
category=tool.category,
arguments=copy.deepcopy(tool.arguments),
steps=copy.deepcopy(selected),
output=f"{{{output_var}}}",
dependencies=[],
system_dependencies=copy.deepcopy(tool.system_dependencies),
visibility=tool.visibility,
input_schema=copy.deepcopy(tool.input_schema),
output_schema=copy.deepcopy(output_schema),
)
def extraction_diff(tool: Tool) -> str:
"""Return a unified-diff preview for a proposed new tool."""
rendered = yaml.safe_dump(tool.to_dict(), sort_keys=False).splitlines(True)
return "".join(difflib.unified_diff(
[], rendered, fromfile="/dev/null", tofile=f"{tool.name}/config.yaml",
))
def _switch_to_existing_tool(main_window, name: str) -> None:
"""Remove the creation page before opening an existing tool editor."""
main_window.close_tool_builder()
@ -35,6 +117,7 @@ class ToolBuilderPage(QWidget):
self.original_name = tool_name
self._tool = None
self._flow_widget = None # Lazy-loaded
self._guidance_worker = None
self._setup_ui()
@ -104,6 +187,44 @@ class ToolBuilderPage(QWidget):
left_layout.addWidget(info_box)
# Reuse guidance is advisory and never changes a draft automatically.
guidance_box = QGroupBox("Reuse & Discovery")
guidance_layout = QVBoxLayout(guidance_box)
self.guidance_summary = QLabel(
"CmdForge can check for similar tools, reusable step sequences, "
"and frequent local pipelines."
)
self.guidance_summary.setWordWrap(True)
guidance_layout.addWidget(self.guidance_summary)
self.guidance_list = QListWidget()
self.guidance_list.setMaximumHeight(130)
self.guidance_list.currentItemChanged.connect(
self._guidance_selection_changed
)
guidance_layout.addWidget(self.guidance_list)
guidance_buttons = QHBoxLayout()
self.btn_refresh_guidance = QPushButton("Refresh")
self.btn_refresh_guidance.clicked.connect(self._refresh_guidance)
guidance_buttons.addWidget(self.btn_refresh_guidance)
self.btn_open_guidance = QPushButton("Open Existing")
self.btn_open_guidance.setObjectName("secondary")
self.btn_open_guidance.clicked.connect(self._open_guidance_tool)
self.btn_open_guidance.setEnabled(False)
guidance_buttons.addWidget(self.btn_open_guidance)
self.btn_extract_guidance = QPushButton("Extract as New Tool")
self.btn_extract_guidance.clicked.connect(self._extract_guidance)
self.btn_extract_guidance.setEnabled(False)
guidance_buttons.addWidget(self.btn_extract_guidance)
guidance_layout.addLayout(guidance_buttons)
left_layout.addWidget(guidance_box)
self._guidance_timer = QTimer(self)
self._guidance_timer.setSingleShot(True)
self._guidance_timer.setInterval(700)
self._guidance_timer.timeout.connect(self._refresh_guidance)
self.name_input.textChanged.connect(self._schedule_guidance)
self.desc_input.textChanged.connect(self._schedule_guidance)
# Arguments group
args_box = QGroupBox()
args_layout = QVBoxLayout(args_box)
@ -404,6 +525,175 @@ class ToolBuilderPage(QWidget):
layout.addWidget(splitter, 1)
def _current_draft(self) -> Tool:
"""Materialize the current form without saving it."""
current = self._tool
return Tool(
name=self.name_input.text().strip() or "untitled-tool",
description=self.desc_input.text().strip(),
category=self.category_combo.currentText() or "Other",
arguments=copy.deepcopy(current.arguments) if current else [],
steps=copy.deepcopy(current.steps) if current else [],
output=self.output_input.toPlainText().strip() or "{response}",
dependencies=list(current.dependencies) if current else [],
system_dependencies=(
copy.deepcopy(current.system_dependencies) if current else []
),
visibility=current.visibility if current else "public",
input_schema=copy.deepcopy(current.input_schema) if current else None,
output_schema=copy.deepcopy(current.output_schema) if current else None,
)
def _schedule_guidance(self):
self._guidance_timer.start()
def _add_guidance_item(self, text: str, data: dict) -> None:
item = QListWidgetItem(text)
item.setData(Qt.UserRole, data)
self.guidance_list.addItem(item)
def _refresh_guidance(self):
"""Refresh deterministic local guidance, then search remotely."""
from ...preflight import analyze_tool
from ...usage import get_suggestions
draft = self._current_draft()
self.guidance_list.clear()
report = analyze_tool(
draft,
check_local_dependencies=False,
include_contract_tests=False,
check_reuse=True,
)
for opportunity in report.reuse_opportunities:
self._add_guidance_item(
f"Reusable: {opportunity.get('detail', 'exact sequence')}",
opportunity,
)
for suggestion in get_suggestions():
tools = suggestion["tools"]
self._add_guidance_item(
f"Frequent pipeline ({suggestion['count']}x): " + " | ".join(tools),
{"type": "usage_pipeline", **suggestion},
)
query = (draft.description or draft.name).strip()
if not query or query == "untitled-tool":
self._finish_guidance(query, [])
return
if self._guidance_worker and self._guidance_worker.isRunning():
# The pending result is ignored if the query has since changed.
self.guidance_summary.setText("Checking registry…")
return
self.guidance_summary.setText("Checking registry…")
self.btn_refresh_guidance.setEnabled(False)
self._guidance_worker = SimilarToolsWorker(query, self)
self._guidance_worker.results_ready.connect(self._finish_guidance)
self._guidance_worker.start()
def _finish_guidance(self, query: str, results: list):
current_query = (
self.desc_input.text().strip() or self.name_input.text().strip()
)
if query == current_query:
seen = set()
for result in results:
name = result.get("name", "")
owner = result.get("owner", "")
reference = f"{owner}/{name}" if owner else name
if reference and reference not in seen:
seen.add(reference)
score = result.get("quality_score")
quality = f" [{score}/100]" if score is not None else ""
self._add_guidance_item(
f"Registry: {reference}{quality} — {result.get('description', '')}",
{"type": "registry_similar", "tool": reference},
)
self.btn_refresh_guidance.setEnabled(True)
count = self.guidance_list.count()
self.guidance_summary.setText(
f"{count} advisory suggestion{'s' if count != 1 else ''}. "
"Nothing changes until you confirm an action."
if count else "No evidenced reuse opportunities found."
)
self._guidance_selection_changed(self.guidance_list.currentItem())
if query != current_query:
self._guidance_timer.start(0)
def _guidance_selection_changed(self, current, previous=None):
data = current.data(Qt.UserRole) if current else {}
kind = data.get("type")
local_name = data.get("tool", "")
self.btn_open_guidance.setEnabled(
kind == "duplicate_sequence" and bool(local_name)
)
self.btn_extract_guidance.setEnabled(
kind in {"repeated_sequence", "usage_pipeline"}
)
def _open_guidance_tool(self):
item = self.guidance_list.currentItem()
data = item.data(Qt.UserRole) if item else {}
if data.get("type") == "duplicate_sequence" and data.get("tool"):
_switch_to_existing_tool(self.main_window, data["tool"])
def _extract_guidance(self):
"""Preview and explicitly save an evidenced reusable tool."""
item = self.guidance_list.currentItem()
data = item.data(Qt.UserRole) if item else {}
kind = data.get("type")
if kind not in {"repeated_sequence", "usage_pipeline"}:
return
suggested = (
"-".join(data.get("tools", []))
if kind == "usage_pipeline" else f"{self.name_input.text().strip()}-shared"
)
name, accepted = QInputDialog.getText(
self, "Extract Reusable Tool", "New tool name:", text=suggested
)
name = name.strip()
if not accepted or not name:
return
valid, error = validate_tool_name(name)
if not valid:
QMessageBox.warning(self, "Validation", error)
return
if tool_exists(name):
QMessageBox.warning(self, "Tool Already Exists", f"'{name}' already exists.")
return
try:
if kind == "usage_pipeline":
from ...usage import build_composite_tool
extracted = build_composite_tool(name, data["tools"])
else:
extracted = build_extracted_tool(self._current_draft(), data, name)
except (KeyError, TypeError, ValueError) as exc:
QMessageBox.warning(self, "Cannot Extract", str(exc))
return
preview = QMessageBox(self)
preview.setIcon(QMessageBox.Information)
preview.setWindowTitle("Review Extraction")
preview.setText(
f"Create '{name}' as a separate tool? Your current draft will not be modified."
)
preview.setDetailedText(extraction_diff(extracted))
create_button = preview.addButton("Create Tool", QMessageBox.AcceptRole)
preview.addButton(QMessageBox.Cancel)
preview.exec()
if preview.clickedButton() is not create_button:
return
try:
save_tool(extracted)
except (OSError, ValueError) as exc:
QMessageBox.critical(self, "Extraction Failed", str(exc))
return
QMessageBox.information(
self, "Tool Created", f"Created '{name}'. The current draft is unchanged."
)
self._populate_deps_combo()
self._refresh_guidance()
def _set_view_mode(self, mode: int):
"""Switch between list (0) and flow (1) views."""
if mode == 1 and self._flow_widget is None:

33
src/cmdforge/improvement.py
View File

@ -63,7 +63,9 @@ class ImprovementReport:
}
def generate_improvements(tool: Tool) -> ImprovementReport:
def generate_improvements(
tool: Tool, scrutiny_report: Optional[dict] = None
) -> ImprovementReport:
"""Analyze a tool and generate improvement suggestions.
Categories:
@ -77,6 +79,35 @@ def generate_improvements(tool: Tool) -> ImprovementReport:
_check_efficiency(tool, report)
_check_transparency(tool, report)
# Preserve the registry scrutiny engine's evidence rather than running a
# disconnected approximation. Only actionable warnings/failures become
# suggestions; passing checks are deliberately omitted.
for finding in (scrutiny_report or {}).get("findings", []):
result = finding.get("result")
if result not in ("warning", "fail"):
continue
check = str(finding.get("check", "scrutiny")).lower()
category = next(
(value for value in ("honesty", "efficiency", "transparency")
if value in check),
"transparency",
)
suggestion = finding.get("suggestion") or (
"Review this finding and make the behavior explicit."
)
candidate = ImprovementSuggestion(
category=category,
severity="high" if result == "fail" else "medium",
title=f"Scrutiny: {finding.get('check', 'review required')}",
description=finding.get("message", "Scrutiny identified an issue."),
location=finding.get("location") or "tool",
suggested=suggestion,
)
marker = (candidate.category, candidate.title, candidate.location)
existing = {(s.category, s.title, s.location) for s in report.suggestions}
if marker not in existing:
report.suggestions.append(candidate)
return report

207
src/cmdforge/integrity.py
View File

@ -1,24 +1,27 @@
"""Transitive integrity verification (M9.4).
"""Content-addressable, transitive integrity verification for CmdForge tools."""
Content-addressable tool identity: tool = hash(tool definition + all dep hashes).
Extends lockfile to include transitive integrity chain.
"""
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional
from typing import Dict, List, Optional, Set
from .tool import Tool, load_tool
from .tool import Tool, ToolStep, load_tool
def _sha256(value: object) -> str:
encoded = json.dumps(
value, sort_keys=True, separators=(",", ":"), ensure_ascii=False
).encode("utf-8")
return f"sha256:{hashlib.sha256(encoded).hexdigest()}"
@dataclass
class IntegrityNode:
"""One node in the integrity chain."""
name: str
version: str = ""
definition_hash: str = ""
content_hash: str = ""
dependencies: List[str] = field(default_factory=list)
dependency_hashes: Dict[str, str] = field(default_factory=dict)
@ -27,6 +30,7 @@ class IntegrityNode:
return {
"name": self.name,
"version": self.version,
"definition_hash": self.definition_hash,
"hash": self.content_hash,
"dependencies": self.dependencies,
"dependency_hashes": self.dependency_hashes,
@ -35,99 +39,142 @@ class IntegrityNode:
@dataclass
class IntegrityChain:
"""Full transitive integrity chain for a tool."""
root: IntegrityNode
nodes: Dict[str, IntegrityNode] = field(default_factory=dict)
errors: List[str] = field(default_factory=list)
@property
def is_valid(self) -> bool:
"""Check all dependency hashes match."""
if self.errors or self.root.name not in self.nodes:
return False
for name, node in self.nodes.items():
if not node.definition_hash or not node.content_hash:
return False
if sorted(node.dependencies) != sorted(node.dependency_hashes):
return False
for dep_name in node.dependencies:
expected_hash = node.dependency_hashes.get(dep_name)
if expected_hash == "unresolved":
# Unresolved dependencies don't invalidate the chain
continue
if dep_name not in self.nodes:
return False
dep_node = self.nodes[dep_name]
if expected_hash and expected_hash != dep_node.content_hash:
dep = self.nodes.get(dep_name)
if dep is None or node.dependency_hashes[dep_name] != dep.content_hash:
return False
expected = _identity_hash(node.definition_hash, node.dependency_hashes)
if not _constant_time_equal(expected, node.content_hash):
return False
return True
def to_dict(self) -> dict:
return {
"root": self.root.to_dict(),
"nodes": {k: v.to_dict() for k, v in self.nodes.items()},
"nodes": {key: value.to_dict() for key, value in self.nodes.items()},
"errors": list(self.errors),
"valid": self.is_valid,
}
def _constant_time_equal(left: str, right: str) -> bool:
import hmac
return hmac.compare_digest(left, right)
def compute_tool_hash(tool: Tool) -> str:
"""Compute a content hash for a tool definition."""
tool_dict = tool.to_dict()
# Remove path (not part of content identity)
tool_dict.pop("path", None)
content = json.dumps(tool_dict, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()[:16]
"""Return the full SHA-256 hash of the canonical tool definition."""
data = tool.to_dict()
data.pop("path", None)
# Registry bookkeeping is not executable tool content.
for key in ("registry_hash", "registry_status", "registry_feedback"):
data.pop(key, None)
return _sha256(data)
def build_integrity_chain(tool: Tool, max_depth: int = 10) -> IntegrityChain:
"""Build a transitive integrity chain for a tool.
def _identity_hash(definition_hash: str, dependency_hashes: Dict[str, str]) -> str:
"""Hash a definition together with the identities of every direct dep."""
return _sha256({
"definition": definition_hash,
"dependencies": dict(sorted(dependency_hashes.items())),
})
Traverses all ToolStep dependencies and computes hashes for each.
"""
chain = IntegrityChain(root=_build_node(tool))
chain.nodes[tool.name] = chain.root
_traverse_deps(tool, chain, depth=0, max_depth=max_depth)
def compute_content_identity(
definition_hash: str, dependency_hashes: Dict[str, str]
) -> str:
"""Return the canonical identity used by integrity chains and lockfiles."""
return _identity_hash(definition_hash, dependency_hashes)
def _dependency_names(tool: Tool) -> List[str]:
names = list(tool.dependencies)
names.extend(step.tool for step in tool.steps if isinstance(step, ToolStep))
return sorted(set(names))
def build_integrity_chain(tool: Tool, max_depth: int = 50) -> IntegrityChain:
"""Resolve dependencies bottom-up and compute a Merkle-style identity."""
placeholder = IntegrityNode(tool.name, tool.version or "")
chain = IntegrityChain(root=placeholder)
visiting: Set[str] = set()
def visit(current: Tool, depth: int) -> Optional[IntegrityNode]:
name = current.name
if name in visiting:
chain.errors.append(f"Dependency cycle detected at {name}")
return None
if name in chain.nodes:
return chain.nodes[name]
if depth > max_depth:
chain.errors.append(f"Maximum dependency depth exceeded at {name}")
return None
visiting.add(name)
dep_hashes: Dict[str, str] = {}
dependencies = _dependency_names(current)
for dep_name in dependencies:
from .resolver import ToolSpec
from .semver import matches_constraint
dep_spec = ToolSpec.parse(dep_name)
dep_tool = load_tool(dep_spec.full_name)
if dep_tool is None:
chain.errors.append(f"Unresolved dependency: {dep_name}")
continue
if dep_spec.version and not matches_constraint(
dep_tool.version or "", dep_spec.version
):
chain.errors.append(
f"Dependency version mismatch: {dep_name} "
f"resolved to {dep_tool.version or 'unknown'}"
)
continue
dep_node = visit(dep_tool, depth + 1)
if dep_node is not None:
dep_hashes[dep_name] = dep_node.content_hash
visiting.remove(name)
definition_hash = compute_tool_hash(current)
node = IntegrityNode(
name=name,
version=current.version or "",
definition_hash=definition_hash,
content_hash=_identity_hash(definition_hash, dep_hashes),
dependencies=dependencies,
dependency_hashes=dep_hashes,
)
chain.nodes[name] = node
return node
root = visit(tool, 0)
if root is not None:
chain.root = root
return chain
def _build_node(tool: Tool) -> IntegrityNode:
"""Build an integrity node from a tool."""
deps = []
dep_hashes = {}
for step in tool.steps:
if hasattr(step, "tool") and step.tool:
deps.append(step.tool)
# Resolve dependency hashes
for dep_name in deps:
dep_tool = load_tool(dep_name)
if dep_tool:
dep_hashes[dep_name] = compute_tool_hash(dep_tool)
else:
dep_hashes[dep_name] = "unresolved"
return IntegrityNode(
name=tool.name,
version=tool.version or "",
content_hash=compute_tool_hash(tool),
dependencies=deps,
dependency_hashes=dep_hashes,
def verify_integrity(chain: IntegrityChain, *, verify_installed: bool = False) -> bool:
"""Verify chain structure, optionally rebuilding it from installed tools."""
if not chain.is_valid:
return False
if not verify_installed:
return True
root_tool = load_tool(chain.root.name)
if root_tool is None:
return False
rebuilt = build_integrity_chain(root_tool)
return rebuilt.is_valid and _constant_time_equal(
rebuilt.root.content_hash, chain.root.content_hash
)
def _traverse_deps(tool: Tool, chain: IntegrityChain, depth: int, max_depth: int):
"""Recursively traverse dependencies and add to chain."""
if depth >= max_depth:
return
for step in tool.steps:
if hasattr(step, "tool") and step.tool:
dep_name = step.tool
if dep_name in chain.nodes:
continue
dep_tool = load_tool(dep_name)
if not dep_tool:
continue
node = _build_node(dep_tool)
chain.nodes[dep_name] = node
_traverse_deps(dep_tool, chain, depth + 1, max_depth)
def verify_integrity(chain: IntegrityChain) -> bool:
"""Verify that all hashes in the chain are consistent."""
return chain.is_valid

64
src/cmdforge/lockfile.py
View File

@ -26,6 +26,8 @@ class LockedPackage:
direct: bool # True if in manifest
required_by: List[str] = field(default_factory=list) # Parent packages
path: Optional[str] = None # Relative path for local tools
content_hash: str = "" # Definition plus dependency identities
dependency_hashes: Dict[str, str] = field(default_factory=dict)
@property
def owner(self) -> str:
@ -145,7 +147,9 @@ class Lockfile:
source=pkg_data.get("source", "registry"),
direct=pkg_data.get("direct", False),
required_by=pkg_data.get("required_by", []),
path=pkg_data.get("path")
path=pkg_data.get("path"),
content_hash=pkg_data.get("content_hash", ""),
dependency_hashes=pkg_data.get("dependency_hashes", {})
)
return cls(
@ -180,6 +184,10 @@ class Lockfile:
pkg_dict["required_by"] = pkg.required_by
if pkg.path:
pkg_dict["path"] = pkg.path
if pkg.content_hash:
pkg_dict["content_hash"] = pkg.content_hash
if pkg.dependency_hashes:
pkg_dict["dependency_hashes"] = dict(sorted(pkg.dependency_hashes.items()))
d["packages"][name] = pkg_dict
return d
@ -259,6 +267,34 @@ def generate_lockfile(
)
lock.packages[qualified_name] = pkg
from .integrity import compute_content_identity
visiting = set()
def identity_for(package_name: str) -> str:
pkg = lock.packages.get(package_name)
node = graph.nodes.get(package_name)
if pkg is None or node is None or not pkg.integrity:
return ""
if pkg.content_hash:
return pkg.content_hash
if package_name in visiting:
return ""
visiting.add(package_name)
dependencies = {}
for child_name in sorted(node.children):
child_hash = identity_for(child_name)
if not child_hash:
visiting.remove(package_name)
return ""
dependencies[child_name] = child_hash
visiting.remove(package_name)
pkg.dependency_hashes = dependencies
pkg.content_hash = compute_content_identity(pkg.integrity, dependencies)
return pkg.content_hash
for package_name in sorted(lock.packages):
identity_for(package_name)
return lock
@ -368,4 +404,30 @@ def verify_lockfile(
except Exception as e:
errors.append(f"{name}: could not verify integrity ({e})")
if locked.content_hash:
from .integrity import compute_content_identity
actual_dependencies = {}
missing_dependencies = []
for dep_name, expected_hash in locked.dependency_hashes.items():
dep = lock.packages.get(dep_name)
if dep is None or not dep.content_hash:
missing_dependencies.append(dep_name)
elif dep.content_hash != expected_hash:
errors.append(
f"{name}: dependency identity mismatch ({dep_name})"
)
else:
actual_dependencies[dep_name] = dep.content_hash
if missing_dependencies:
errors.append(
f"{name}: unresolved integrity dependencies: "
+ ", ".join(missing_dependencies)
)
elif locked.integrity:
actual_identity = compute_content_identity(
locked.integrity, actual_dependencies
)
if actual_identity != locked.content_hash:
errors.append(f"{name}: transitive content identity mismatch")
return errors

143
src/cmdforge/mcp_client.py
View File

@ -13,6 +13,7 @@ from dataclasses import dataclass, field
from datetime import timedelta
from pathlib import Path
from typing import Any, Awaitable, Callable, Dict, List, Optional, TypeVar
from urllib.parse import urlparse
import yaml
@ -20,7 +21,7 @@ MCP_CONFIG_FILE = Path.home() / ".cmdforge" / "mcp.yaml"
MCP_CONFIG_VERSION = 1
MCP_DEPTH_ENV = "CMDFORGE_MCP_DEPTH"
RESULT_MODES = ("auto", "structured", "content", "text")
SUPPORTED_TRANSPORTS = ("stdio",)
SUPPORTED_TRANSPORTS = ("stdio", "streamable-http")
DEFAULT_INHERITED_ENV = (
"PATH",
"HOME",
@ -95,13 +96,15 @@ def _glob_match(pattern: str, name: str) -> bool:
class McpServerConfig:
"""Configuration for one MCP server.
M7.2 intentionally supports stdio only. ``approved`` records explicit
user consent to execute the configured local command.
``approved`` records explicit user consent to execute a local command or
connect to the configured HTTP endpoint.
"""
name: str
transport: str = "stdio"
command: Optional[str] = None
url: Optional[str] = None
headers: Dict[str, str] = field(default_factory=dict)
args: List[str] = field(default_factory=list)
cwd: Optional[str] = None
env: Dict[str, str] = field(default_factory=dict)
@ -116,10 +119,25 @@ class McpServerConfig:
if self.transport not in SUPPORTED_TRANSPORTS:
raise ValueError(
f"Unsupported MCP transport '{self.transport}'. "
f"Supported in M7.2: {', '.join(SUPPORTED_TRANSPORTS)}"
f"Supported transports: {', '.join(SUPPORTED_TRANSPORTS)}"
)
if not isinstance(self.command, str) or not self.command.strip():
raise ValueError(f"MCP server '{self.name}' requires a command")
if self.transport == "stdio":
if not isinstance(self.command, str) or not self.command.strip():
raise ValueError(f"MCP server '{self.name}' requires a command")
if self.url is not None:
raise ValueError(f"MCP stdio server '{self.name}' cannot define url")
if self.headers:
raise ValueError(f"MCP stdio server '{self.name}' cannot define headers")
else:
if self.command is not None:
raise ValueError(
f"MCP streamable-http server '{self.name}' cannot define command"
)
_validate_remote_url(self.name, self.url)
if self.args or self.cwd or self.env:
raise ValueError(
f"MCP streamable-http server '{self.name}' cannot define args, cwd, or env"
)
if not isinstance(self.args, list) or not all(isinstance(arg, str) for arg in self.args):
raise ValueError(f"MCP server '{self.name}' args must be a list of strings")
if not isinstance(self.env, dict) or not all(
@ -131,6 +149,15 @@ class McpServerConfig:
raise ValueError(
f"MCP server '{self.name}' env must map valid variable names to strings"
)
if not isinstance(self.headers, dict) or not all(
isinstance(key, str) and key.strip()
and "\n" not in key and "\r" not in key
and isinstance(value, str) and "\n" not in value and "\r" not in value
for key, value in self.headers.items()
):
raise ValueError(
f"MCP server '{self.name}' headers must map safe names to strings"
)
if not isinstance(self.inherit_env, list) or not all(
isinstance(name, str) and _ENV_NAME.fullmatch(name)
for name in self.inherit_env
@ -155,6 +182,8 @@ def _fingerprint(cfg: McpServerConfig) -> str:
[
cfg.transport,
cfg.command,
cfg.url,
cfg.headers,
cfg.args,
cfg.cwd,
cfg.env,
@ -203,6 +232,36 @@ def _build_server_env(cfg: McpServerConfig) -> Dict[str, str]:
return environment
def _is_loopback_host(hostname: Optional[str]) -> bool:
if not hostname:
return False
if hostname.lower() == "localhost":
return True
try:
import ipaddress
return ipaddress.ip_address(hostname).is_loopback
except ValueError:
return False
def _validate_remote_url(name: str, url: Optional[str]) -> None:
if not isinstance(url, str) or not url.strip():
raise ValueError(f"MCP streamable-http server '{name}' requires a url")
parsed = urlparse(url)
if parsed.scheme not in ("http", "https") or not parsed.hostname:
raise ValueError(f"MCP server '{name}' has an invalid HTTP URL")
if parsed.username or parsed.password:
raise ValueError("MCP URLs must not contain credentials; use headers")
if parsed.scheme != "https" and not _is_loopback_host(parsed.hostname):
raise ValueError(
f"Remote MCP server '{name}' must use HTTPS; HTTP is allowed only for localhost"
)
def _build_http_headers(cfg: McpServerConfig) -> Dict[str, str]:
return {name: _expand_env_value(value) for name, value in cfg.headers.items()}
def _load_mcp_document() -> dict:
if not MCP_CONFIG_FILE.exists():
return {}
@ -229,6 +288,8 @@ def load_mcp_config() -> List[McpServerConfig]:
name=name,
transport=raw.get("transport", "stdio"),
command=raw.get("command"),
url=raw.get("url"),
headers=raw.get("headers", {}),
args=raw.get("args", []),
cwd=raw.get("cwd"),
env=raw.get("env", {}),
@ -276,9 +337,14 @@ def save_mcp_config(servers: List[McpServerConfig]) -> None:
def _server_to_dict(server: McpServerConfig) -> dict:
data: dict = {
"transport": server.transport,
"command": server.command,
"approved": server.approved,
}
if server.command:
data["command"] = server.command
if server.url:
data["url"] = server.url
if server.headers:
data["headers"] = server.headers
if server.args:
data["args"] = server.args
if server.cwd:
@ -329,8 +395,8 @@ def _serialize_tool(tool: Any) -> Dict[str, Any]:
class McpClientManager:
"""Invocation-scoped MCP configuration and schema manager.
SDK sessions are intentionally scoped to each discovery or tool operation
in M7.2. This guarantees subprocess cleanup while the manager reuses loaded
SDK sessions are intentionally scoped to each discovery or tool operation.
This guarantees transport cleanup while the manager reuses loaded
configuration and discovered schemas across all MCP steps in one tool run.
"""
@ -368,7 +434,10 @@ class McpClientManager:
raise KeyError(f"MCP server '{server_name}' not configured. Add it to {MCP_CONFIG_FILE}")
cfg.validate()
if not cfg.approved:
command = " ".join([cfg.command or ""] + cfg.args)
command = (
" ".join([cfg.command or ""] + cfg.args)
if cfg.transport == "stdio" else cfg.url or ""
)
raise PermissionError(
f"MCP server '{server_name}' is not approved to execute: {_sanitize(command)}. "
"Re-add it with 'cmdforge mcp add ...' or set approved: true after review."
@ -401,7 +470,7 @@ class McpClientManager:
seen_cursors.add(cursor)
return tools
tools = _run_stdio_operation(cfg, list_all_tools, "discover tools")
tools = _run_operation(cfg, list_all_tools, "discover tools")
self._schemas[server_name] = {tool["name"]: tool for tool in tools}
self._schema_fingerprints[server_name] = fingerprint
return tools
@ -440,7 +509,57 @@ class McpClientManager:
)
return _normalize_result(result, result_mode)
return _run_stdio_operation(cfg, call, f"call {tool_name}")
return _run_operation(cfg, call, f"call {tool_name}")
def _run_operation(
cfg: McpServerConfig,
operation: Callable[[Any], Awaitable[T]],
operation_name: str,
) -> T:
if cfg.transport == "stdio":
return _run_stdio_operation(cfg, operation, operation_name)
return _run_http_operation(cfg, operation, operation_name)
def _run_http_operation(
cfg: McpServerConfig,
operation: Callable[[Any], Awaitable[T]],
operation_name: str,
) -> T:
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
import httpx
resolved_headers = _build_http_headers(cfg)
async def run() -> T:
timeout = httpx.Timeout(cfg.timeout)
async with httpx.AsyncClient(
headers=resolved_headers, timeout=timeout, follow_redirects=False
) as client:
async with streamable_http_client(
cfg.url, http_client=client
) as (read, write, _):
async with ClientSession(
read, write,
read_timeout_seconds=timedelta(seconds=cfg.timeout),
) as session:
await session.initialize()
return await operation(session)
try:
return asyncio.run(asyncio.wait_for(run(), timeout=cfg.timeout))
except (asyncio.TimeoutError, TimeoutError) as exc:
raise TimeoutError(
f"MCP {operation_name} on '{cfg.name}' timed out after {cfg.timeout}s"
) from exc
except Exception as exc:
secrets = list(resolved_headers.values())
raise RuntimeError(
f"MCP {operation_name} on '{cfg.name}' failed: "
f"{_sanitize(str(exc), secrets)}"
) from exc
def _run_stdio_operation(

106
src/cmdforge/mcp_server.py
View File

@ -1,14 +1,18 @@
"""MCP server support: expose CmdForge tools as MCP tools."""
import inspect
import hmac
import os
import sys
from typing import Annotated, Any, Literal
from typing import Annotated, Any, Literal, Optional
from urllib.parse import urlparse
from .mcp_client import (
MCP_DEPTH_ENV,
McpServeConfig,
_MCP_CALL_DEPTH,
_expand_env_value,
_is_loopback_host,
_require_mcp_sdk,
load_mcp_serve_config,
)
@ -87,16 +91,70 @@ def _build_tool_schema(tool) -> dict:
}
def serve(transport: str = "stdio") -> None:
class _StaticTokenVerifier:
def __init__(self, token: str):
self._token = token
async def verify_token(self, token: str):
if not hmac.compare_digest(token, self._token):
return None
from mcp.server.auth.provider import AccessToken
return AccessToken(
token=token, client_id="cmdforge-mcp-client", scopes=[],
subject="cmdforge-mcp",
)
def _validate_http_server_options(
host: str,
port: int,
allowed_origins: list[str],
auth_token: Optional[str],
external_url: Optional[str],
) -> tuple[list[str], Optional[str]]:
if isinstance(port, bool) or not isinstance(port, int) or not 1 <= port <= 65535:
raise ValueError("MCP server port must be between 1 and 65535")
local = _is_loopback_host(host)
resolved_token = _expand_env_value(auth_token) if auth_token else None
if not local:
parsed = urlparse(external_url or "")
if parsed.scheme != "https" or not parsed.hostname:
raise ValueError(
"Non-local MCP serving requires --external-url https://... "
"behind a TLS-terminating proxy"
)
if not resolved_token:
raise ValueError("Non-local MCP serving requires --auth-token")
origins = allowed_origins or (
[f"http://localhost:{port}", f"http://127.0.0.1:{port}"]
if local else [f"{urlparse(external_url).scheme}://{urlparse(external_url).netloc}"]
)
for origin in origins:
parsed = urlparse(origin)
if parsed.scheme not in ("http", "https") or not parsed.netloc or parsed.path not in ("", "/"):
raise ValueError(f"Invalid allowed origin: {origin}")
if not local and parsed.scheme != "https":
raise ValueError("Non-local MCP origins must use HTTPS")
return origins, resolved_token
def serve(
transport: str = "stdio",
*,
host: str = "127.0.0.1",
port: int = 8000,
allowed_origins: Optional[list[str]] = None,
auth_token: Optional[str] = None,
external_url: Optional[str] = None,
) -> None:
"""Start CmdForge as an MCP server on the given transport.
Currently only stdio transport is supported (M7.3).
HTTP defaults to loopback and enables SDK DNS-rebinding/origin protection.
Non-local binding requires bearer auth and an explicit public HTTPS URL;
TLS is expected to terminate at the reverse proxy represented by that URL.
"""
if transport != "stdio":
raise NotImplementedError(
f"Unsupported MCP server transport '{transport}'. "
f"M7.3 supports stdio only."
)
if transport not in ("stdio", "streamable-http"):
raise ValueError(f"Unsupported MCP server transport '{transport}'")
_require_mcp_sdk()
from mcp.server.fastmcp import FastMCP
@ -104,7 +162,37 @@ def serve(transport: str = "stdio") -> None:
serve_config = load_mcp_serve_config()
from .tool import list_tools, load_tool
server = FastMCP("CmdForge")
server_options = {}
if transport == "streamable-http":
origins, resolved_token = _validate_http_server_options(
host, port, allowed_origins or [], auth_token, external_url
)
from mcp.server.transport_security import TransportSecuritySettings
allowed_hosts = [host, f"{host}:{port}", "localhost", f"localhost:{port}"]
if external_url:
external_host = urlparse(external_url).netloc
allowed_hosts.extend([external_host, urlparse(external_url).hostname or ""])
server_options.update({
"host": host,
"port": port,
"transport_security": TransportSecuritySettings(
enable_dns_rebinding_protection=True,
allowed_hosts=[value for value in allowed_hosts if value],
allowed_origins=origins,
),
})
if resolved_token:
from mcp.server.auth.settings import AuthSettings
issuer = external_url or f"http://{host}:{port}"
server_options.update({
"token_verifier": _StaticTokenVerifier(resolved_token),
"auth": AuthSettings(
issuer_url=issuer,
resource_server_url=f"{issuer.rstrip('/')}/mcp",
required_scopes=[],
),
})
server = FastMCP("CmdForge", **server_options)
tool_names = list_tools()
exposed_count = 0

138
src/cmdforge/prompt_optimizer.py
View File

@ -5,7 +5,7 @@ contract conformance tests, and surfaces the highest-scoring variation.
"""
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from typing import Callable, Dict, List, Optional
from .tool import Tool, PromptStep
@ -37,6 +37,8 @@ class OptimizationResult:
scores: Dict[int, int] = field(default_factory=dict) # variation_index -> score
best_index: Optional[int] = None
baseline_score: int = 0
evaluation_kind: str = "structural"
note: str = ""
@property
def best(self) -> Optional[PromptVariation]:
@ -55,6 +57,8 @@ class OptimizationResult:
return {
"tool": self.tool_name,
"baseline_score": self.baseline_score,
"evaluation_kind": self.evaluation_kind,
"note": self.note,
"best_index": self.best_index,
"improvement_pct": round(self.improvement, 1),
"best": self.best.to_dict() if self.best else None,
@ -93,7 +97,7 @@ def generate_variations(
original = step.prompt
strategies = _get_strategies(original, count)
strategies = _get_strategies(original, count, provider=provider)
for strategy, variation_text in strategies:
if variation_text and variation_text != original:
@ -107,16 +111,25 @@ def generate_variations(
return variations
def _get_strategies(original: str, count: int) -> List[tuple]:
def _get_strategies(
original: str, count: int, provider: str = "mock"
) -> List[tuple]:
"""Generate variations using different strategies."""
strategies: List[tuple] = []
# Attempt AI-driven rephrase if a real provider is available
rephrased = _call_provider_for_variation(
f"Rephrase this instruction while keeping the same meaning. Output only the rephrased text:\n\n{original}"
)
if rephrased and rephrased != original:
strategies.append(("rephrase", rephrased))
if count <= 0:
return []
# Provider calls are explicit. The default/mock path is deterministic and
# never sends a user's prompt to an external process or service.
if provider != "mock":
rephrased = _call_provider_for_variation(
provider,
"Rephrase this instruction while keeping the same meaning. "
f"Output only the rephrased text:\n\n{original}",
)
if rephrased and rephrased != original:
strategies.append(("rephrase", rephrased))
# Deterministic mock variations
mock_strategies = [
@ -141,17 +154,15 @@ def _truncate(text: str, max_len: int) -> str:
return text[:max_len - 3].rsplit(" ", 1)[0] + "..."
def _call_provider_for_variation(prompt: str) -> Optional[str]:
"""Try to call a provider for variation generation."""
try:
from .providers import call_provider
result = call_provider("opencode-pickle", prompt, timeout=15)
if result.success and result.text:
text = result.text.strip()
if text and len(text) >= 10:
return text
except Exception:
pass
def _call_provider_for_variation(provider: str, prompt: str) -> Optional[str]:
"""Call the provider explicitly selected by the user."""
from .providers import call_provider
result = call_provider(provider, prompt, timeout=15)
if result.success and result.text:
text = result.text.strip()
if text and len(text) >= 10:
return text
return None
@ -159,6 +170,7 @@ def optimize_tool(
tool: Tool,
count: int = 5,
provider: str = "mock",
evaluator: Optional[Callable[[Tool], float]] = None,
) -> OptimizationResult:
"""Run prompt optimization for a tool.
@ -175,11 +187,17 @@ def optimize_tool(
"""
from .contract_testing import run_contract_tests
result = OptimizationResult(tool_name=tool.name)
result = OptimizationResult(
tool_name=tool.name,
evaluation_kind="behavioral" if evaluator else "structural",
)
# Baseline score
baseline = run_contract_tests(tool)
result.baseline_score = _count_passed(baseline)
if evaluator:
result.baseline_score = evaluator(tool)
else:
baseline = run_contract_tests(tool)
result.baseline_score = _count_passed(baseline)
# Generate variations
result.variations = generate_variations(tool, count=count, provider=provider)
@ -198,14 +216,26 @@ def optimize_tool(
step.prompt = variation.variation
try:
test_result = run_contract_tests(tool_copy)
result.scores[idx] = _count_passed(test_result)
if evaluator:
result.scores[idx] = evaluator(tool_copy)
else:
test_result = run_contract_tests(tool_copy)
result.scores[idx] = _count_passed(test_result)
except Exception:
result.scores[idx] = 0
# Find best
if result.scores:
result.best_index = max(result.scores, key=lambda k: result.scores[k])
if evaluator and result.scores:
candidate = max(result.scores, key=lambda k: result.scores[k])
if result.scores[candidate] > result.baseline_score:
result.best_index = candidate
else:
result.note = "No variation improved on the behavioral baseline."
elif result.scores:
result.note = (
"Structural conformance cannot compare prompt semantics; "
"no best variation was selected."
)
return result
@ -214,3 +244,57 @@ def _count_passed(test_result) -> int:
"""Count passed tests in a ConformanceReport."""
return sum(1 for r in getattr(test_result, "results", [])
if getattr(r, "state", "") == "passed")
def load_behavioral_evaluator(path, provider_override: Optional[str] = None):
"""Load explicit behavioral cases and return a semantic score function.
The JSON file must contain a list of cases. Each case accepts ``input`` and
optional ``args``, plus exactly one of ``expected`` (exact value) or
``contains`` (substring). Supplying this file is explicit authorization to
execute the candidate tools and their configured providers.
"""
import json
from pathlib import Path
cases = json.loads(Path(path).read_text(encoding="utf-8"))
if not isinstance(cases, list) or not cases:
raise ValueError("Behavior test file must contain a non-empty JSON list")
for index, case in enumerate(cases):
if not isinstance(case, dict):
raise ValueError(f"Behavior case {index} must be an object")
assertions = [key for key in ("expected", "contains") if key in case]
if len(assertions) != 1:
raise ValueError(
f"Behavior case {index} needs exactly one of expected or contains"
)
if not isinstance(case.get("args", {}), dict):
raise ValueError(f"Behavior case {index} args must be an object")
def evaluate(tool: Tool) -> float:
from .runner import run_tool
passed = 0
for case in cases:
output, exit_code = run_tool(
tool,
input_text=str(case.get("input", "")),
custom_args=case.get("args", {}),
provider_override=provider_override,
dry_run=False,
verbose=False,
)
if exit_code != 0:
continue
if "expected" in case:
try:
actual = json.loads(output)
except (TypeError, json.JSONDecodeError):
actual = output
if actual == case["expected"]:
passed += 1
elif str(case["contains"]) in output:
passed += 1
return passed
return evaluate

319
src/cmdforge/registry/app.py
View File

@ -1140,6 +1140,25 @@ def create_app() -> Flask:
except (AttributeError, TypeError, ValueError):
audit_stale = True
contributor_rows = query_all(
g.db,
"""SELECT p.slug, c.contribution FROM tool_contributors c
JOIN publishers p ON p.id = c.publisher_id
WHERE c.tool_id = ? ORDER BY p.slug""",
[row["id"]],
)
approved_improvement = query_one(
g.db,
"""SELECT id FROM improvement_submissions
WHERE applied_tool_id = ? LIMIT 1""",
[row["id"]],
)
badges = []
if row.get("attestation_json"):
badges.append("verified")
if approved_improvement:
badges.extend(["optimized", "community-reviewed"])
payload = {
"owner": row["owner"],
"name": row["name"],
@ -1164,6 +1183,11 @@ def create_app() -> Flask:
"quality": json.loads(latest_audit["quality_json"]) if latest_audit else None,
"audit_evaluated_at": latest_audit["evaluated_at"] if latest_audit else None,
"audit_stale": audit_stale,
"contributors": [
{"name": item["slug"], "contribution": item["contribution"]}
for item in contributor_rows
],
"badges": badges,
}
response = jsonify({"data": payload})
response.headers["Cache-Control"] = "max-age=60"
@ -1313,6 +1337,18 @@ def create_app() -> Flask:
except Exception:
g.db.rollback()
publisher = query_one(
g.db, "SELECT signing_public_key FROM publishers WHERE id = ?",
[row["publisher_id"]],
)
attestation = None
if row.get("attestation_json"):
try:
attestation = json.loads(row["attestation_json"])
except (TypeError, json.JSONDecodeError):
return error_response(
"INVALID_ATTESTATION", "Stored release attestation is corrupt", 500
)
response = jsonify({
"data": {
"owner": row["owner"],
@ -1321,12 +1357,172 @@ def create_app() -> Flask:
"config": row["config_yaml"],
"readme": row["readme"] or "",
"config_hash": row.get("config_hash") or "",
"content_hash": row.get("content_hash") or "",
"dependency_hashes": json.loads(
row.get("dependency_hashes_json") or "{}"
),
"defaults": row.get("defaults") or "",
"attestation": attestation,
"signing_public_key": (
publisher["signing_public_key"] if publisher else ""
) or "",
}
})
response.headers["Cache-Control"] = "max-age=3600, immutable"
return response
@app.route("/api/v1/tools/by-content-hash/<path:content_hash>", methods=["GET"])
def tool_by_content_hash(content_hash: str) -> Response:
"""Resolve an approved public release by its transitive identity."""
if not re.fullmatch(r"sha256:[0-9a-f]{64}", content_hash):
return error_response("VALIDATION_ERROR", "Invalid content hash", 400)
row = query_one(
g.db,
"""SELECT owner, name, version, description, category, tags,
downloads, published_at
FROM tools WHERE content_hash = ? AND visibility = 'public'
AND moderation_status = 'approved'
ORDER BY published_at DESC LIMIT 1""",
[content_hash],
)
if not row:
return error_response("TOOL_NOT_FOUND", "Content hash not found", 404)
result = dict(row)
result["tags"] = json.loads(result.get("tags") or "[]")
return jsonify({"data": result})
@app.route(
"/api/v1/tools/<owner>/<name>/<version>/improvements", methods=["POST"]
)
@require_token
def submit_improvement(owner: str, name: str, version: str) -> Response:
"""Submit and automatically test a prompt/code improvement."""
row = query_one(
g.db,
"""SELECT id, config_yaml, visibility, owner FROM tools
WHERE owner = ? AND name = ? AND version = ?
AND moderation_status = 'approved'""",
[owner, name, version],
)
if not row:
return error_response("TOOL_NOT_FOUND", "Tool release not found", 404)
if (
row["visibility"] != "public"
and g.current_publisher["slug"] != row["owner"]
and g.current_publisher.get("role") not in ("moderator", "admin")
):
return error_response("TOOL_NOT_FOUND", "Tool release not found", 404)
payload = request.get_json(silent=True) or {}
try:
from ..community import create_submission, run_submission_tests
from ..tool import Tool
tool = Tool.from_dict(yaml.safe_load(row["config_yaml"]) or {})
submission = create_submission(
tool,
int(payload.get("step_index", -1)),
str(payload.get("proposed", "")),
g.current_publisher["slug"],
str(payload.get("rationale", "")),
)
if not submission.proposed.strip():
raise ValueError("Proposed content must not be empty")
test_result = run_submission_tests(submission, tool)
except (KeyError, TypeError, ValueError) as exc:
return error_response("VALIDATION_ERROR", str(exc), 400)
if not test_result.get("passed_for_review"):
return error_response(
"IMPROVEMENT_TEST_FAILED",
test_result.get("error") or "Proposed change did not pass automated tests",
400,
details={"test_result": test_result},
)
cursor = g.db.execute(
"""INSERT INTO improvement_submissions (
tool_id, submitter_id, step_index, step_type, original,
proposed, rationale, status, test_result_json, submitted_at
) VALUES (?, ?, ?, ?, ?, ?, ?, 'tested', ?, ?)""",
[
row["id"], g.current_publisher["id"], submission.step_index,
submission.step_type, submission.original, submission.proposed,
submission.rationale, json.dumps(test_result, sort_keys=True),
submission.submitted_at,
],
)
g.db.commit()
return jsonify({
"data": {"id": cursor.lastrowid, **submission.to_dict()}
}), 201
@app.route("/api/v1/improvements/<int:submission_id>", methods=["PATCH"])
@require_token
def review_improvement(submission_id: int) -> Response:
"""Allow the tool owner/moderator to review a tested submission."""
row = query_one(
g.db,
"""SELECT s.*, t.owner FROM improvement_submissions s
JOIN tools t ON t.id = s.tool_id WHERE s.id = ?""",
[submission_id],
)
if not row:
return error_response("NOT_FOUND", "Improvement not found", 404)
if (
g.current_publisher["slug"] != row["owner"]
and g.current_publisher.get("role") not in ("moderator", "admin")
):
return error_response("FORBIDDEN", "Owner or moderator required", 403)
decision = (request.get_json(silent=True) or {}).get("decision")
if decision not in ("approve", "reject", "request_changes"):
return error_response("VALIDATION_ERROR", "Invalid decision", 400)
status = "approved" if decision == "approve" else decision
now = datetime.now(timezone.utc).isoformat()
notes = (request.get_json(silent=True) or {}).get("notes", "")
g.db.execute(
"""UPDATE improvement_submissions SET status = ?, reviewer_id = ?,
review_notes = ?, reviewed_at = ? WHERE id = ?""",
[status, g.current_publisher["id"], notes, now, submission_id],
)
g.db.commit()
return jsonify({
"data": {
"id": submission_id, "status": status,
"ready_to_apply": status == "approved",
}
})
@app.route("/api/v1/me/signing-key", methods=["PUT"])
@require_token
def set_signing_key() -> Response:
"""Register an Ed25519 public key as the publisher trust anchor."""
public_key = (request.get_json(silent=True) or {}).get("public_key", "")
try:
import base64
raw = base64.b64decode(public_key, validate=True)
if len(raw) != 32:
raise ValueError
except (ValueError, TypeError):
return error_response(
"VALIDATION_ERROR", "public_key must be a base64 Ed25519 key", 400
)
existing_key = query_one(
g.db, "SELECT signing_public_key FROM publishers WHERE id = ?",
[g.current_publisher["id"]],
)
current_key = (
existing_key["signing_public_key"] if existing_key else ""
) or ""
if current_key and current_key != public_key:
return error_response(
"SIGNING_KEY_EXISTS",
"Signing key rotation requires administrator recovery",
409,
)
g.db.execute(
"UPDATE publishers SET signing_public_key = ? WHERE id = ?",
[public_key, g.current_publisher["id"]],
)
g.db.commit()
return jsonify({"data": {"signing_public_key": public_key}})
@app.route("/api/v1/categories", methods=["GET"])
def list_categories() -> Response:
page, per_page, sort, order, error = parse_pagination("/categories", "name")
@ -2470,6 +2666,8 @@ def create_app() -> Flask:
readme = payload.get("readme") or ""
defaults = payload.get("defaults") or ""
dry_run = bool(payload.get("dry_run"))
attestation_data = payload.get("attestation")
improvement_id = payload.get("improvement_id")
size_resp = validate_payload_size("config", config_text, MAX_CONFIG_BYTES)
if size_resp:
@ -2638,10 +2836,64 @@ def create_app() -> Flask:
# Compute config hash early for idempotency check
config_hash = compute_yaml_hash(config_text)
from ..integrity import compute_content_identity
from ..tool import ToolStep
dependency_names = set(published_tool.dependencies)
dependency_names.update(
step.tool for step in published_tool.steps if isinstance(step, ToolStep)
)
dependency_hashes = {}
for dependency_name in sorted(dependency_names):
from ..resolver import ToolSpec
dep_spec = ToolSpec.parse(dependency_name)
dep_owner = dep_spec.owner or owner
dep_row = resolve_tool(dep_owner, dep_spec.name, dep_spec.version)
if dep_row:
dep_identity = dep_row.get("content_hash") or dep_row.get("config_hash")
if dep_identity:
dependency_hashes[dependency_name] = dep_identity
content_hash = compute_content_identity(config_hash, dependency_hashes)
publisher_record = query_one(
g.db, "SELECT signing_public_key FROM publishers WHERE id = ?",
[g.current_publisher["id"]],
)
signing_public_key = (
publisher_record["signing_public_key"] if publisher_record else ""
) or ""
attestation_json = None
if signing_public_key and not dry_run and not attestation_data:
return error_response(
"ATTESTATION_REQUIRED",
"This publisher has a signing key; releases must be signed",
400,
)
if attestation_data:
try:
from ..attestation import Attestation, verify_attestation
attestation = Attestation.from_dict(attestation_data)
except (KeyError, TypeError, ValueError):
return error_response(
"INVALID_ATTESTATION", "Malformed release attestation", 400
)
if (
attestation.tool_name != name
or attestation.version != version
or attestation.content_hash != content_hash
or attestation.signer != owner
or not signing_public_key
or not verify_attestation(attestation, signing_public_key)
):
return error_response(
"INVALID_ATTESTATION",
"Release attestation does not match this publisher and content",
400,
)
attestation_json = json.dumps(attestation.to_dict(), sort_keys=True)
existing = query_one(
g.db,
"SELECT published_at, config_hash, moderation_status, visibility FROM tools WHERE owner = ? AND name = ? AND version = ?",
"SELECT published_at, config_hash, content_hash, moderation_status, visibility FROM tools WHERE owner = ? AND name = ? AND version = ?",
[owner, name, version],
)
if existing:
@ -2653,6 +2905,7 @@ def create_app() -> Flask:
"name": name,
"version": version,
"config_hash": config_hash,
"content_hash": existing.get("content_hash") or content_hash,
"pr_url": "",
"status": existing["moderation_status"],
"visibility": existing["visibility"],
@ -2667,6 +2920,43 @@ def create_app() -> Flask:
details={"published_at": existing["published_at"]},
)
applied_improvement = None
if improvement_id is not None:
try:
improvement_id = int(improvement_id)
except (TypeError, ValueError):
return error_response("VALIDATION_ERROR", "Invalid improvement_id", 400)
applied_improvement = query_one(
g.db,
"""SELECT s.*, t.owner, t.name FROM improvement_submissions s
JOIN tools t ON t.id = s.tool_id
WHERE s.id = ? AND s.status = 'approved'
AND s.applied_tool_id IS NULL""",
[improvement_id],
)
if (
not applied_improvement
or applied_improvement["owner"] != owner
or applied_improvement["name"] != name
or applied_improvement["step_index"] >= len(published_tool.steps)
):
return error_response(
"VALIDATION_ERROR", "Improvement cannot be applied to this release", 400
)
candidate_step = published_tool.steps[applied_improvement["step_index"]]
candidate_value = (
candidate_step.prompt if applied_improvement["step_type"] == "prompt"
and hasattr(candidate_step, "prompt") else
candidate_step.code if applied_improvement["step_type"] == "code"
and hasattr(candidate_step, "code") else None
)
if candidate_value != applied_improvement["proposed"]:
return error_response(
"VALIDATION_ERROR",
"Published config does not contain the approved improvement",
400,
)
suggestions = {"category": None, "similar_tools": []}
try:
from .categorize import suggest_categories
@ -2714,6 +3004,10 @@ def create_app() -> Flask:
# Check scrutiny decision
if scrutiny_report:
suggestions["scrutiny"] = scrutiny_report
from ..improvement import generate_improvements
suggestions["improvements"] = generate_improvements(
published_tool, scrutiny_report
).to_dict()
if scrutiny_report.get("decision") == "reject":
# Find the failing check for error message
fail_findings = [f for f in scrutiny_report.get("findings", []) if f.get("result") == "fail"]
@ -2749,6 +3043,7 @@ def create_app() -> Flask:
"suggestions": suggestions,
"preflight": preflight_report.to_dict(),
"quality": quality_report,
"content_hash": content_hash,
}
})
@ -2780,9 +3075,10 @@ def create_app() -> Flask:
owner, name, version, description, category, tags, config_yaml, readme,
defaults, publisher_id, deprecated, deprecated_message, replacement, downloads,
scrutiny_status, scrutiny_report, source, source_url, source_json,
config_hash, visibility, moderation_status, forked_from, forked_version,
published_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
config_hash, content_hash, dependency_hashes_json, visibility,
moderation_status, forked_from, forked_version, published_at,
attestation_json
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
[
owner,
@ -2805,14 +3101,28 @@ def create_app() -> Flask:
source_url,
source_json,
config_hash,
content_hash,
json.dumps(dependency_hashes, sort_keys=True),
visibility,
moderation_status,
forked_from,
forked_version,
datetime.now(timezone.utc).isoformat(),
attestation_json,
],
)
tool_id = insert_cursor.lastrowid
if applied_improvement:
g.db.execute(
"UPDATE improvement_submissions SET applied_tool_id = ? WHERE id = ?",
[tool_id, applied_improvement["id"]],
)
g.db.execute(
"""INSERT OR IGNORE INTO tool_contributors
(tool_id, publisher_id, contribution)
VALUES (?, ?, 'improvement')""",
[tool_id, applied_improvement["submitter_id"]],
)
audit = preflight_report.audit_evidence or {}
g.db.execute(
"""
@ -2866,6 +3176,7 @@ def create_app() -> Flask:
"name": name,
"version": version,
"config_hash": config_hash,
"content_hash": content_hash,
"pr_url": "",
"status": moderation_status,
"visibility": visibility,

28
src/cmdforge/registry/auditing.py
View File

@ -26,22 +26,32 @@ def registry_dependency_findings(conn, tool: Tool, owner: str) -> Dict[str, List
missing: List[str] = []
deprecated: List[str] = []
for reference in sorted(references):
if reference == tool.name or reference == f"{owner}/{tool.name}":
from ..resolver import ToolSpec
from ..semver import matches_constraint
spec = ToolSpec.parse(reference)
if spec.name == tool.name and (spec.owner in (None, owner)):
continue
if "/" in reference:
dep_owner, dep_name = reference.split("/", 1)
candidates = [(dep_owner, dep_name)]
if spec.owner:
candidates = [(spec.owner, spec.name)]
else:
candidates = [(owner, reference), ("official", reference)]
candidates = [(owner, spec.name), ("official", spec.name)]
row = None
for dep_owner, dep_name in candidates:
row = conn.execute(
rows = conn.execute(
"""
SELECT deprecated, replacement FROM tools
WHERE owner = ? AND name = ? ORDER BY id DESC LIMIT 1
SELECT version, deprecated, replacement FROM tools
WHERE owner = ? AND name = ? ORDER BY id DESC
""",
[dep_owner, dep_name],
).fetchone()
).fetchall()
row = next(
(
candidate for candidate in rows
if not spec.version
or matches_constraint(candidate["version"], spec.version)
),
None,
)
if row is not None:
break
if row is None:

46
src/cmdforge/registry/db.py
View File

@ -27,6 +27,7 @@ CREATE TABLE IF NOT EXISTS publishers (
ban_reason TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
,signing_public_key TEXT
);
CREATE TABLE IF NOT EXISTS api_tokens (
@ -61,6 +62,8 @@ CREATE TABLE IF NOT EXISTS tools (
source_url TEXT,
source_json TEXT,
config_hash TEXT,
content_hash TEXT,
dependency_hashes_json TEXT,
visibility TEXT DEFAULT 'public',
moderation_status TEXT DEFAULT 'pending',
moderation_note TEXT,
@ -69,6 +72,7 @@ CREATE TABLE IF NOT EXISTS tools (
forked_from TEXT,
forked_version TEXT,
published_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
attestation_json TEXT,
UNIQUE(owner, name, version)
);
@ -94,6 +98,31 @@ CREATE TABLE IF NOT EXISTS tool_audits (
CREATE INDEX IF NOT EXISTS idx_tool_audits_tool_time
ON tool_audits(tool_id, evaluated_at DESC, id DESC);
CREATE TABLE IF NOT EXISTS improvement_submissions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
tool_id INTEGER NOT NULL REFERENCES tools(id) ON DELETE CASCADE,
submitter_id INTEGER NOT NULL REFERENCES publishers(id),
step_index INTEGER NOT NULL,
step_type TEXT NOT NULL,
original TEXT NOT NULL,
proposed TEXT NOT NULL,
rationale TEXT,
status TEXT NOT NULL DEFAULT 'tested',
test_result_json TEXT NOT NULL,
reviewer_id INTEGER REFERENCES publishers(id),
review_notes TEXT,
submitted_at TIMESTAMP NOT NULL,
reviewed_at TIMESTAMP
,applied_tool_id INTEGER REFERENCES tools(id)
);
CREATE TABLE IF NOT EXISTS tool_contributors (
tool_id INTEGER NOT NULL REFERENCES tools(id) ON DELETE CASCADE,
publisher_id INTEGER NOT NULL REFERENCES publishers(id),
contribution TEXT NOT NULL,
PRIMARY KEY (tool_id, publisher_id, contribution)
);
CREATE VIRTUAL TABLE IF NOT EXISTS tools_fts USING fts5(
name, description, tags, readme,
content='tools',
@ -489,6 +518,8 @@ def migrate_db(conn: sqlite3.Connection) -> None:
("source_url", "TEXT", "NULL"),
("source_json", "TEXT", "NULL"),
("config_hash", "TEXT", "NULL"),
("content_hash", "TEXT", "NULL"),
("dependency_hashes_json", "TEXT", "NULL"),
("visibility", "TEXT", "'public'"),
("moderation_status", "TEXT", "'pending'"),
("moderation_note", "TEXT", "NULL"),
@ -497,6 +528,7 @@ def migrate_db(conn: sqlite3.Connection) -> None:
("forked_from", "TEXT", "NULL"),
("forked_version", "TEXT", "NULL"),
("defaults", "TEXT", "NULL"),
("attestation_json", "TEXT", "NULL"),
]
for col_name, col_type, default in tools_migrations:
@ -514,6 +546,7 @@ def migrate_db(conn: sqlite3.Connection) -> None:
("banned_at", "TIMESTAMP", "NULL"),
("banned_by", "TEXT", "NULL"),
("ban_reason", "TEXT", "NULL"),
("signing_public_key", "TEXT", "NULL"),
]
for col_name, col_type, default in publishers_migrations:
@ -540,6 +573,18 @@ def migrate_db(conn: sqlite3.Connection) -> None:
except sqlite3.OperationalError:
pass
cursor = conn.execute("PRAGMA table_info(improvement_submissions)")
improvement_cols = {row[1] for row in cursor.fetchall()}
if improvement_cols and "applied_tool_id" not in improvement_cols:
try:
conn.execute(
"ALTER TABLE improvement_submissions "
"ADD COLUMN applied_tool_id INTEGER REFERENCES tools(id)"
)
conn.commit()
except sqlite3.OperationalError:
pass
# Grandfather existing tools: set moderation_status to 'approved' for tools that have NULL
# This ensures existing tools remain visible after migration (one-time migration)
# Note: Only applies to NULL, NOT to 'pending' - pending tools need manual review
@ -573,6 +618,7 @@ def migrate_db(conn: sqlite3.Connection) -> None:
conn.execute("CREATE INDEX IF NOT EXISTS idx_tools_owner ON tools(owner)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_tools_moderation ON tools(moderation_status, visibility)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_tools_hash ON tools(config_hash)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_tools_content_hash ON tools(content_hash)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_publishers_role ON publishers(role)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_publishers_banned ON publishers(banned)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_audit_log_target ON audit_log(target_type, target_id)")

60
src/cmdforge/registry_client.py
View File

@ -116,6 +116,10 @@ class DownloadResult:
readme: str = ""
config_hash: str = "" # Registry hash for integrity verification
defaults: str = "" # Default settings YAML content
attestation: Optional[Dict[str, Any]] = None
signing_public_key: str = ""
content_hash: str = ""
dependency_hashes: Dict[str, str] = field(default_factory=dict)
class RegistryClient:
@ -562,8 +566,56 @@ class RegistryClient:
readme=data.get("readme", ""),
config_hash=data.get("config_hash", ""),
defaults=data.get("defaults", "")
,attestation=data.get("attestation")
,signing_public_key=data.get("signing_public_key", "")
,content_hash=data.get("content_hash", "")
,dependency_hashes=data.get("dependency_hashes", {})
)
def get_tool_by_content_hash(self, content_hash: str) -> ToolInfo:
response = self._request("GET", f"/tools/by-content-hash/{content_hash}")
if response.status_code != 200:
self._handle_error_response(response)
return ToolInfo.from_dict(response.json().get("data", {}))
def set_signing_public_key(self, public_key: str) -> Dict[str, Any]:
"""Register the public key used to verify this publisher's releases."""
response = self._request(
"PUT", "/me/signing-key",
json_data={"public_key": public_key}, require_auth=True,
)
if response.status_code != 200:
self._handle_error_response(response)
return response.json().get("data", {})
def submit_improvement(
self, owner: str, name: str, version: str, step_index: int,
proposed: str, rationale: str = "",
) -> Dict[str, Any]:
response = self._request(
"POST", f"/tools/{owner}/{name}/{version}/improvements",
json_data={
"step_index": step_index, "proposed": proposed,
"rationale": rationale,
},
require_auth=True,
)
if response.status_code != 201:
self._handle_error_response(response)
return response.json().get("data", {})
def review_improvement(
self, submission_id: int, decision: str, notes: str = ""
) -> Dict[str, Any]:
response = self._request(
"PATCH", f"/improvements/{submission_id}",
json_data={"decision": decision, "notes": notes},
require_auth=True,
)
if response.status_code != 200:
self._handle_error_response(response)
return response.json().get("data", {})
def get_categories(self) -> List[Dict[str, Any]]:
"""
Get list of tool categories.
@ -623,7 +675,9 @@ class RegistryClient:
defaults: str = "",
dry_run: bool = False,
visibility: str = "public",
owner: str = ""
owner: str = "",
attestation: Optional[Dict[str, Any]] = None,
improvement_id: Optional[int] = None,
) -> Dict[str, Any]:
"""
Publish a tool to the registry.
@ -649,6 +703,10 @@ class RegistryClient:
payload["defaults"] = defaults
if owner:
payload["owner"] = owner
if attestation:
payload["attestation"] = attestation
if improvement_id is not None:
payload["improvement_id"] = improvement_id
response = self._request(
"POST",

33
src/cmdforge/resolver.py
View File

@ -354,7 +354,10 @@ class ToolResolver:
config_yaml=result.config_yaml,
readme=result.readme,
config_hash=result.config_hash,
defaults=result.defaults
defaults=result.defaults,
attestation=result.attestation,
signing_public_key=result.signing_public_key,
content_hash=result.content_hash,
)
if self.verbose:
@ -383,7 +386,10 @@ class ToolResolver:
config_yaml: str,
readme: str = "",
config_hash: str = "",
defaults: str = ""
defaults: str = "",
attestation: Optional[dict] = None,
signing_public_key: str = "",
content_hash: str = "",
) -> ResolvedTool:
"""Install a tool fetched from registry to global directory."""
# Verify hash if provided
@ -396,6 +402,24 @@ class ToolResolver:
f"got {computed_hash[:20]}... - content may have been tampered with"
)
if attestation:
from .attestation import Attestation, verify_attestation
try:
signed = Attestation.from_dict(attestation)
except (KeyError, TypeError, ValueError) as exc:
raise RuntimeError("Malformed release attestation") from exc
if (
not signing_public_key
or signed.tool_name != name
or signed.version != version
or signed.content_hash != content_hash
or signed.signer != owner
or not verify_attestation(signed, signing_public_key)
):
raise RuntimeError(
f"Invalid release attestation for {owner}/{name}@{version}"
)
# Create directory structure
tool_dir = TOOLS_DIR / owner / name
tool_dir.mkdir(parents=True, exist_ok=True)
@ -628,7 +652,10 @@ def install_from_registry(spec: str, version: Optional[str] = None) -> ResolvedT
config_yaml=result.config_yaml,
readme=result.readme,
config_hash=result.config_hash,
defaults=result.defaults
defaults=result.defaults,
attestation=result.attestation,
signing_public_key=result.signing_public_key,
content_hash=result.content_hash,
)

8
src/cmdforge/runner.py
View File

@ -1234,6 +1234,14 @@ def main():
else:
print(output)
if exit_code == 0 and not args.dry_run:
try:
from .usage import record_invocation
record_invocation(resolved.full_name)
except (OSError, ValueError):
# Usage discovery is optional and must never break tool execution.
pass
sys.exit(exit_code)

43
src/cmdforge/signing.py Normal file
View File

@ -0,0 +1,43 @@
"""Local publisher signing-key storage."""
from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Optional, Tuple
from .attestation import generate_keypair
from .config import CONFIG_DIR
SIGNING_KEY_FILE = CONFIG_DIR / "release-signing-key.json"
def load_signing_key(path: Path = SIGNING_KEY_FILE) -> Optional[Tuple[str, str]]:
if not path.exists():
return None
data = json.loads(path.read_text(encoding="utf-8"))
private_key = data.get("private_key", "")
public_key = data.get("public_key", "")
if not private_key or not public_key:
raise ValueError("Signing key file is incomplete")
return private_key, public_key
def initialize_signing_key(
path: Path = SIGNING_KEY_FILE, *, overwrite: bool = False
) -> Tuple[str, str]:
if path.exists() and not overwrite:
existing = load_signing_key(path)
if existing is None:
raise ValueError("Signing key file could not be loaded")
return existing
private_key, public_key = generate_keypair()
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
path.write_text(json.dumps({
"algorithm": "ed25519",
"private_key": private_key,
"public_key": public_key,
}, indent=2) + "\n", encoding="utf-8")
os.chmod(path, 0o600)
return private_key, public_key

202
src/cmdforge/usage.py Normal file
View File

@ -0,0 +1,202 @@
"""Opt-in, local-only pipeline usage discovery.
Only tool names, anonymous pipe inode numbers, counts, and timestamps are
stored. Inputs, outputs, arguments, environment, and working directories are
never recorded.
"""
from __future__ import annotations
import json
import os
import stat
import tempfile
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, Iterator, List, Optional
from .config import CONFIG_DIR
USAGE_FILE = CONFIG_DIR / "usage.json"
USAGE_LOCK_FILE = CONFIG_DIR / ".usage.lock"
USAGE_VERSION = 1
EVENT_TTL_SECONDS = 120
MAX_EVENTS = 200
DEFAULT_THRESHOLD = 3
def _empty_document(enabled: bool = False) -> dict:
return {
"version": USAGE_VERSION,
"enabled": enabled,
"events": [],
"pipelines": {},
}
@contextmanager
def _locked() -> Iterator[None]:
CONFIG_DIR.mkdir(parents=True, exist_ok=True, mode=0o700)
with open(USAGE_LOCK_FILE, "a+", encoding="utf-8") as lock:
os.chmod(USAGE_LOCK_FILE, 0o600)
try:
import fcntl
fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
except ImportError:
pass
try:
yield
finally:
try:
import fcntl
fcntl.flock(lock.fileno(), fcntl.LOCK_UN)
except ImportError:
pass
def _load_unlocked() -> dict:
if not USAGE_FILE.exists():
return _empty_document()
try:
data = json.loads(USAGE_FILE.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return _empty_document()
if not isinstance(data, dict) or data.get("version") != USAGE_VERSION:
return _empty_document()
return data
def _save_unlocked(data: dict) -> None:
CONFIG_DIR.mkdir(parents=True, exist_ok=True, mode=0o700)
temp_path = None
try:
with tempfile.NamedTemporaryFile(
"w", encoding="utf-8", dir=CONFIG_DIR,
prefix=".usage-", suffix=".tmp", delete=False,
) as handle:
temp_path = Path(handle.name)
os.chmod(temp_path, 0o600)
json.dump(data, handle, indent=2, sort_keys=True)
handle.write("\n")
handle.flush()
os.fsync(handle.fileno())
os.replace(temp_path, USAGE_FILE)
os.chmod(USAGE_FILE, 0o600)
finally:
if temp_path and temp_path.exists():
temp_path.unlink()
def is_enabled() -> bool:
with _locked():
return bool(_load_unlocked().get("enabled", False))
def set_enabled(enabled: bool) -> None:
with _locked():
data = _load_unlocked()
data["enabled"] = bool(enabled)
if not enabled:
data["events"] = []
_save_unlocked(data)
def clear_usage() -> None:
with _locked():
enabled = bool(_load_unlocked().get("enabled", False))
_save_unlocked(_empty_document(enabled=enabled))
def _pipe_inode(stream) -> Optional[int]:
try:
info = os.fstat(stream.fileno())
except (AttributeError, OSError, ValueError):
return None
return info.st_ino if stat.S_ISFIFO(info.st_mode) else None
def record_invocation(tool_name: str, stdin=None, stdout=None) -> None:
"""Record a successful invocation when it participates in a shell pipe."""
import sys
stdin = stdin or sys.stdin
stdout = stdout or sys.stdout
input_pipe = _pipe_inode(stdin)
output_pipe = _pipe_inode(stdout)
if input_pipe is None and output_pipe is None:
return
now = datetime.now(timezone.utc)
timestamp = now.timestamp()
with _locked():
data = _load_unlocked()
if not data.get("enabled", False):
return
events = [
event for event in data.get("events", [])
if timestamp - float(event.get("time", 0)) <= EVENT_TTL_SECONDS
]
discovered = set()
for event in events:
if input_pipe is not None and event.get("stdout_pipe") == input_pipe:
discovered.add((event["tool"], tool_name))
if output_pipe is not None and event.get("stdin_pipe") == output_pipe:
discovered.add((tool_name, event["tool"]))
pipelines = data.setdefault("pipelines", {})
for first, second in discovered:
if first == second:
continue
key = json.dumps([first, second], separators=(",", ":"))
current = pipelines.setdefault(key, {"tools": [first, second], "count": 0})
current["count"] = int(current.get("count", 0)) + 1
current["last_seen"] = now.isoformat()
events.append({
"tool": tool_name,
"time": timestamp,
"stdin_pipe": input_pipe,
"stdout_pipe": output_pipe,
})
data["events"] = events[-MAX_EVENTS:]
_save_unlocked(data)
def get_suggestions(threshold: int = DEFAULT_THRESHOLD) -> List[Dict]:
with _locked():
data = _load_unlocked()
if not data.get("enabled", False):
return []
suggestions = [
{
"tools": list(item.get("tools", [])),
"count": int(item.get("count", 0)),
"last_seen": item.get("last_seen", ""),
}
for item in data.get("pipelines", {}).values()
if int(item.get("count", 0)) >= threshold
and len(item.get("tools", [])) >= 2
]
return sorted(suggestions, key=lambda item: (-item["count"], item["tools"]))
def build_composite_tool(name: str, tools: List[str]):
"""Build, but do not save, a Tool that composes the suggested pipeline."""
from .tool import Tool, ToolStep
if len(tools) < 2:
raise ValueError("A composite requires at least two tools")
steps = []
previous = "input"
for index, tool_name in enumerate(tools, start=1):
output_var = f"pipeline_{index}"
steps.append(ToolStep(
tool=tool_name,
input_template="{input}" if previous == "input" else f"{{{previous}}}",
output_var=output_var,
))
previous = output_var
return Tool(
name=name,
description="Composite pipeline: " + " | ".join(tools),
dependencies=list(dict.fromkeys(tools)),
steps=steps,
output=f"{{{previous}}}",
)

56
tests/test_attestation.py
View File

@ -2,44 +2,70 @@
from cmdforge.attestation import (
Attestation,
generate_keypair,
sign_tool,
verify_attestation,
verify_content_hash,
verify_trusted_attestation,
)
from cmdforge.tool import Tool
class TestSignTool:
def test_creates_attestation(self):
att = sign_tool("mytool", "1.0.0", "abc123", "alice", "secret-key")
private, public = generate_keypair()
att = sign_tool("mytool", "1.0.0", "abc123", "alice", private)
assert att.tool_name == "mytool"
assert att.version == "1.0.0"
assert att.content_hash == "abc123"
assert att.signer == "alice"
assert len(att.signature) == 64 # SHA256 hex
assert att.algorithm == "hmac-sha256"
assert len(att.signature) == 88
assert att.algorithm == "ed25519"
assert verify_attestation(att, public)
def test_different_keys_different_signatures(self):
att1 = sign_tool("tool", "1.0.0", "hash", "alice", "key1")
att2 = sign_tool("tool", "1.0.0", "hash", "alice", "key2")
key1, _ = generate_keypair()
key2, _ = generate_keypair()
att1 = sign_tool("tool", "1.0.0", "hash", "alice", key1)
att2 = sign_tool("tool", "1.0.0", "hash", "alice", key2)
assert att1.signature != att2.signature
class TestVerifyAttestation:
def test_valid_signature(self):
att = sign_tool("mytool", "1.0.0", "abc123", "alice", "secret-key")
assert verify_attestation(att, "secret-key")
private, public = generate_keypair()
att = sign_tool("mytool", "1.0.0", "abc123", "alice", private)
assert verify_attestation(att, public)
def test_wrong_key_fails(self):
att = sign_tool("mytool", "1.0.0", "abc123", "alice", "secret-key")
assert not verify_attestation(att, "wrong-key")
private, _ = generate_keypair()
_, wrong_public = generate_keypair()
att = sign_tool("mytool", "1.0.0", "abc123", "alice", private)
assert not verify_attestation(att, wrong_public)
def test_tampered_content_fails(self):
att = sign_tool("mytool", "1.0.0", "abc123", "alice", "secret-key")
att.content_hash = "tampered"
assert not verify_attestation(att, "secret-key")
from dataclasses import replace
private, public = generate_keypair()
att = sign_tool("mytool", "1.0.0", "abc123", "alice", private)
att = replace(att, content_hash="tampered")
assert not verify_attestation(att, public)
def test_tampered_signer_fails(self):
att = sign_tool("mytool", "1.0.0", "abc123", "alice", "secret-key")
att.signer = "eve"
assert not verify_attestation(att, "secret-key")
from dataclasses import replace
private, public = generate_keypair()
att = sign_tool("mytool", "1.0.0", "abc123", "alice", private)
att = replace(att, signer="eve")
assert not verify_attestation(att, public)
def test_timestamp_and_algorithm_are_signed(self):
from dataclasses import replace
private, public = generate_keypair()
att = sign_tool("tool", "1.0.0", "hash", "alice", private)
assert not verify_attestation(replace(att, signed_at="tomorrow"), public)
assert not verify_attestation(replace(att, algorithm="hmac-sha256"), public)
def test_requires_trusted_publisher_mapping(self):
private, public = generate_keypair()
att = sign_tool("tool", "1.0.0", "hash", "alice", private)
assert verify_trusted_attestation(att, {"alice": public})
assert not verify_trusted_attestation(att, {"mallory": public})

73
tests/test_cli.py
View File

@ -650,3 +650,76 @@ def test_switch_to_existing_tool_closes_creation_page_first():
call.close_tool_builder(),
call.open_tool_builder("existing"),
]
def test_guided_extraction_builds_new_tool_without_mutating_draft():
pytest.importorskip("PySide6")
from cmdforge.gui.pages.tool_builder_page import (
build_extracted_tool,
extraction_diff,
)
schema = {"type": "string"}
repeated = [
PromptStep(
prompt="First {input}", provider="mock", output_var="first",
output_schema=schema,
),
PromptStep(
prompt="Second {first}", provider="mock", output_var="second",
output_schema=schema,
),
]
draft = Tool(
name="draft", steps=repeated + [
PromptStep(
prompt="First {input}", provider="mock", output_var="first_copy",
output_schema=schema,
),
PromptStep(
prompt="Second {first}", provider="mock", output_var="second_copy",
output_schema=schema,
),
], input_schema=schema,
)
before = draft.to_dict()
extracted = build_extracted_tool(
draft,
{"type": "repeated_sequence", "locations": [1, 3], "length": 2},
"draft-shared",
)
assert len(extracted.steps) == 2
assert extracted.output == "{second}"
assert extracted.output_schema == schema
assert draft.to_dict() == before
assert "+++ draft-shared/config.yaml" in extraction_diff(extracted)
def test_guided_extraction_rejects_hidden_external_step_dependency():
pytest.importorskip("PySide6")
from cmdforge.gui.pages.tool_builder_page import build_extracted_tool
schema = {"type": "string"}
draft = Tool(name="draft", steps=[
PromptStep(
prompt="Produce", provider="mock", output_var="earlier",
output_schema=schema,
),
PromptStep(
prompt="Use {earlier}", provider="mock", output_var="one",
output_schema=schema,
),
PromptStep(
prompt="Finish {one}", provider="mock", output_var="two",
output_schema=schema,
),
])
with pytest.raises(ValueError, match="outside its boundary"):
build_extracted_tool(
draft,
{"type": "repeated_sequence", "locations": [2, 4], "length": 2},
"draft-shared",
)

140
tests/test_collection_api.py
View File

@ -339,6 +339,146 @@ class TestPublishPreflightEndpoint:
assert response.status_code == 200
@flask_required
class TestM9RegistryTrustAndCommunity:
def test_signed_publish_download_and_content_lookup(
self, client, auth_headers
):
from cmdforge.attestation import (
Attestation, generate_keypair, sign_tool, verify_attestation,
)
from cmdforge.registry.db import connect_db
private_key, public_key = generate_keypair()
response = client.put(
"/api/v1/me/signing-key", headers=auth_headers,
json={"public_key": public_key},
)
assert response.status_code == 200
config = "name: signed-tool\nversion: 1.0.0\noutput: stable\n"
preflight = client.post(
"/api/v1/tools", headers=auth_headers,
json={"config": config, "dry_run": True},
)
assert preflight.status_code == 200
content_hash = preflight.get_json()["data"]["content_hash"]
attestation = sign_tool(
"signed-tool", "1.0.0", content_hash, "testuser", private_key
)
published = client.post(
"/api/v1/tools", headers=auth_headers,
json={"config": config, "attestation": attestation.to_dict()},
)
assert published.status_code == 201
assert published.get_json()["data"]["content_hash"] == content_hash
conn = connect_db()
try:
conn.execute(
"UPDATE tools SET moderation_status = 'approved' "
"WHERE owner = 'testuser' AND name = 'signed-tool'"
)
conn.commit()
finally:
conn.close()
downloaded = client.get(
"/api/v1/tools/testuser/signed-tool/download?install=false"
)
assert downloaded.status_code == 200
data = downloaded.get_json()["data"]
assert data["content_hash"] == content_hash
assert verify_attestation(Attestation.from_dict(data["attestation"]), public_key)
lookup = client.get(f"/api/v1/tools/by-content-hash/{content_hash}")
assert lookup.status_code == 200
assert lookup.get_json()["data"]["name"] == "signed-tool"
def test_registered_key_requires_valid_signature(self, client, auth_headers):
from cmdforge.attestation import generate_keypair
_, public_key = generate_keypair()
assert client.put(
"/api/v1/me/signing-key", headers=auth_headers,
json={"public_key": public_key},
).status_code == 200
response = client.post(
"/api/v1/tools", headers=auth_headers,
json={"config": "name: unsigned-tool\nversion: 1.0.0\n"},
)
assert response.status_code == 400
assert response.get_json()["error"]["code"] == "ATTESTATION_REQUIRED"
def test_improvement_is_tested_reviewed_and_credited(
self, client, auth_headers
):
from cmdforge.registry.db import connect_db
config = (
"name: improvable\nversion: 1.0.0\noutput: '{result}'\n"
"input_schema:\n type: string\noutput_schema:\n type: string\n"
"steps:\n - type: prompt\n prompt: 'Summarize: {input}'\n"
" provider: mock\n output_var: result\n"
)
published = client.post(
"/api/v1/tools", headers=auth_headers, json={"config": config}
)
assert published.status_code == 201
conn = connect_db()
try:
conn.execute(
"UPDATE tools SET moderation_status = 'approved' "
"WHERE owner = 'testuser' AND name = 'improvable'"
)
conn.commit()
finally:
conn.close()
submitted = client.post(
"/api/v1/tools/testuser/improvable/1.0.0/improvements",
headers=auth_headers,
json={
"step_index": 0,
"proposed": "Summarize the input accurately and concisely: {input}",
"rationale": "Clearer expected behavior",
},
)
assert submitted.status_code == 201
submission_id = submitted.get_json()["data"]["id"]
reviewed = client.patch(
f"/api/v1/improvements/{submission_id}", headers=auth_headers,
json={"decision": "approve", "notes": "Validated"},
)
assert reviewed.status_code == 200
assert reviewed.get_json()["data"]["ready_to_apply"] is True
improved_config = config.replace("version: 1.0.0", "version: 1.0.1").replace(
"Summarize: {input}",
"Summarize the input accurately and concisely: {input}",
)
applied = client.post(
"/api/v1/tools", headers=auth_headers,
json={"config": improved_config, "improvement_id": submission_id},
)
assert applied.status_code == 201
conn = connect_db()
try:
conn.execute(
"UPDATE tools SET moderation_status = 'approved' "
"WHERE owner = 'testuser' AND name = 'improvable' "
"AND version = '1.0.1'"
)
conn.commit()
credited = conn.execute(
"SELECT COUNT(*) AS count FROM tool_contributors"
).fetchone()["count"]
finally:
conn.close()
assert credited == 1
detail = client.get("/api/v1/tools/testuser/improvable?version=1.0.1")
assert detail.status_code == 200
assert detail.get_json()["data"]["badges"] == [
"optimized", "community-reviewed"
]
@flask_required
class TestPostCollectionsEndpoint:
"""Tests for POST /api/v1/collections endpoint."""

4
tests/test_community.py
View File

@ -49,11 +49,11 @@ class TestReviewSubmission:
sub = ImprovementSubmission(
tool_name="test", tool_version="1.0.0", submitter="alice",
step_index=0, step_type="prompt", original="a", proposed="b",
status="tested",
status="tested", test_result={"passed_for_review": True},
)
review = review_submission(sub, "approve", "admin", "Good improvement")
assert review.decision == "approve"
assert sub.status == "approve"
assert sub.status == "approved"
def test_review_pending_rejected(self):
sub = ImprovementSubmission(

38
tests/test_integrity.py
View File

@ -44,12 +44,14 @@ class TestIntegrityChain:
)
chain = build_integrity_chain(tool)
assert "missing-dep" not in chain.nodes
# Chain is still valid because the dependency hash is "unresolved"
# and there's no node to compare against
assert chain.is_valid
assert not chain.is_valid
assert chain.errors == ["Unresolved dependency: missing-dep"]
def test_chain_valid_with_resolved_dep(self, tmp_path):
with patch("cmdforge.tool.TOOLS_DIR", tmp_path / ".cmdforge"):
with (
patch("cmdforge.tool.TOOLS_DIR", tmp_path / ".cmdforge"),
patch("cmdforge.tool.BIN_DIR", tmp_path / "bin"),
):
from cmdforge.tool import save_tool
child = Tool(name="child", version="1.0.0")
@ -67,13 +69,35 @@ class TestIntegrityChain:
assert "child" in chain.nodes
assert chain.is_valid
def test_child_change_changes_root_identity(self, tmp_path):
with (
patch("cmdforge.tool.TOOLS_DIR", tmp_path / ".cmdforge"),
patch("cmdforge.tool.BIN_DIR", tmp_path / "bin"),
):
from cmdforge.tool import save_tool
child = Tool(name="child", version="1.0.0", output="one")
save_tool(child)
parent = Tool(
name="parent", version="1.0.0",
steps=[ToolStep(tool="child", output_var="x")], output="{x}",
)
first = build_integrity_chain(parent).root.content_hash
child.output = "two"
save_tool(child)
second = build_integrity_chain(parent).root.content_hash
assert first != second
class TestVerifyIntegrity:
def test_valid_chain(self):
chain = IntegrityChain(
root=IntegrityNode(name="root", content_hash="abc"),
nodes={"root": IntegrityNode(name="root", content_hash="abc")},
from cmdforge.integrity import compute_content_identity
definition_hash = "sha256:" + "a" * 64
content_hash = compute_content_identity(definition_hash, {})
node = IntegrityNode(
name="root", definition_hash=definition_hash,
content_hash=content_hash,
)
chain = IntegrityChain(root=node, nodes={"root": node})
assert verify_integrity(chain)
def test_tampered_chain(self):

14
tests/test_lockfile.py
View File

@ -259,6 +259,12 @@ class TestGenerateLockfile:
assert "official/text-utils" in lock.packages
assert lock.packages["official/summarize"].direct is True
assert lock.packages["official/text-utils"].direct is False
child_identity = lock.packages["official/text-utils"].content_hash
assert child_identity.startswith("sha256:")
assert lock.packages["official/summarize"].dependency_hashes == {
"official/text-utils": child_identity
}
assert lock.packages["official/summarize"].content_hash.startswith("sha256:")
class TestVerifyLockfile:
@ -319,7 +325,9 @@ class TestLockfileRoundTrip:
integrity="sha256:abc123",
source="registry",
direct=True,
required_by=[]
required_by=[],
content_hash="sha256:root",
dependency_hashes={"official/text-utils": "sha256:child"},
),
"official/text-utils": LockedPackage(
name="official/text-utils",
@ -364,6 +372,10 @@ class TestLockfileRoundTrip:
assert summarize.integrity == "sha256:abc123"
assert summarize.source == "registry"
assert summarize.direct is True
assert summarize.content_hash == "sha256:root"
assert summarize.dependency_hashes == {
"official/text-utils": "sha256:child"
}
text_utils = loaded.packages["official/text-utils"]
assert text_utils.direct is False

92
tests/test_mcp.py
View File

@ -12,6 +12,7 @@ from cmdforge.mcp_client import (
McpServerConfig,
McpClientManager,
_build_server_env,
_build_http_headers,
_normalize_result,
_sanitize,
_fingerprint,
@ -34,14 +35,15 @@ class TestMcpServerConfig:
@pytest.mark.parametrize(
"change, message",
[
({"transport": "streamable-http"}, "Unsupported MCP transport"),
({"transport": "streamable-http"}, "requires a url"),
({"args": "-y"}, "list of strings"),
({"timeout": 0}, "greater than 0"),
({"timeout": True}, "must be a number"),
],
)
def test_validation_rejects_invalid_config(self, change, message):
cfg = McpServerConfig(name="test", command="server", **change)
command = None if change.get("transport") == "streamable-http" else "server"
cfg = McpServerConfig(name="test", command=command, **change)
with pytest.raises(ValueError, match=message):
cfg.validate()
@ -55,6 +57,31 @@ class TestMcpServerConfig:
b = McpServerConfig(name="b", command="cmd", args=["-v"])
assert _fingerprint(a) == _fingerprint(b)
def test_streamable_http_accepts_https(self):
cfg = McpServerConfig(
name="remote", transport="streamable-http",
url="https://example.com/mcp", approved=True,
)
cfg.validate()
def test_streamable_http_allows_loopback_http_only(self):
McpServerConfig(
name="local", transport="streamable-http",
url="http://127.0.0.1:8000/mcp",
).validate()
with pytest.raises(ValueError, match="must use HTTPS"):
McpServerConfig(
name="remote", transport="streamable-http",
url="http://example.com/mcp",
).validate()
def test_streamable_http_rejects_url_credentials(self):
with pytest.raises(ValueError, match="must not contain credentials"):
McpServerConfig(
name="remote", transport="streamable-http",
url="https://user:pass@example.com/mcp",
).validate()
class TestSanitize:
def test_redacts_bearer(self):
@ -85,6 +112,15 @@ class TestEnvironmentIsolation:
)
assert _build_server_env(cfg)["SERVER_TOKEN"] == "allowed"
def test_http_header_reference_is_resolved(self, monkeypatch):
monkeypatch.setenv("MCP_TOKEN", "secret")
cfg = McpServerConfig(
name="remote", transport="streamable-http",
url="https://example.com/mcp",
headers={"Authorization": "Bearer ${MCP_TOKEN}"},
)
assert _build_http_headers(cfg)["Authorization"] == "Bearer secret"
def test_missing_environment_reference_fails(self, monkeypatch):
monkeypatch.delenv("CMDFORGE_MISSING", raising=False)
cfg = McpServerConfig(
@ -122,6 +158,19 @@ class TestMcpConfigPersistence:
def test_load_no_file(self, temp_mcp_file):
assert load_mcp_config() == []
def test_streamable_http_round_trip(self, temp_mcp_file):
save_mcp_config([McpServerConfig(
name="remote", transport="streamable-http",
url="https://example.com/mcp",
headers={"Authorization": "Bearer ${MCP_TOKEN}"},
approved=True,
)])
loaded = load_mcp_config()[0]
assert loaded.transport == "streamable-http"
assert loaded.command is None
assert loaded.url == "https://example.com/mcp"
assert loaded.headers == {"Authorization": "Bearer ${MCP_TOKEN}"}
def test_save_creates_0600_permissions(self, temp_mcp_file):
save_mcp_config([McpServerConfig(name="test", command="echo")])
perms = oct(temp_mcp_file.stat().st_mode & 0o777)
@ -651,6 +700,45 @@ class TestRegisteredMcpTool:
serve()
class TestStreamableHttpServerSafety:
def test_defaults_are_loopback_with_origin_allowlist(self):
from cmdforge.mcp_server import _validate_http_server_options
origins, token = _validate_http_server_options(
"127.0.0.1", 8000, [], None, None
)
assert token is None
assert origins == [
"http://localhost:8000", "http://127.0.0.1:8000"
]
def test_nonlocal_requires_https_and_auth(self):
from cmdforge.mcp_server import _validate_http_server_options
with pytest.raises(ValueError, match="external-url"):
_validate_http_server_options("0.0.0.0", 8000, [], "secret", None)
with pytest.raises(ValueError, match="auth-token"):
_validate_http_server_options(
"0.0.0.0", 8000, [], None, "https://mcp.example.com"
)
def test_nonlocal_infers_https_origin_and_expands_token(self, monkeypatch):
from cmdforge.mcp_server import _validate_http_server_options
monkeypatch.setenv("MCP_AUTH", "secret-value")
origins, token = _validate_http_server_options(
"0.0.0.0", 8000, [], "${MCP_AUTH}",
"https://mcp.example.com",
)
assert origins == ["https://mcp.example.com"]
assert token == "secret-value"
def test_nonlocal_rejects_insecure_origin(self):
from cmdforge.mcp_server import _validate_http_server_options
with pytest.raises(ValueError, match="origins must use HTTPS"):
_validate_http_server_options(
"0.0.0.0", 8000, ["http://example.com"], "secret",
"https://mcp.example.com",
)
class TestCmdForgeMcpServerEndToEnd:
def test_stdio_discovery_and_invocation(self, tmp_path, monkeypatch):
pytest.importorskip("mcp")

38
tests/test_prompt_optimizer.py
View File

@ -23,6 +23,21 @@ class TestTruncate:
class TestGenerateVariations:
def test_mock_provider_never_calls_external_provider(self):
with patch("cmdforge.providers.call_provider") as call:
_get_strategies("Summarize the text", 2, provider="mock")
call.assert_not_called()
def test_selected_provider_is_honored(self):
from cmdforge.providers import ProviderResult
with patch(
"cmdforge.providers.call_provider",
return_value=ProviderResult(text="A materially clearer instruction", success=True),
) as call:
result = _get_strategies("Summarize the text", 1, provider="chosen")
assert result[0][1] == "A materially clearer instruction"
assert call.call_args.args[0] == "chosen"
def test_generates_for_prompt_steps(self, tmp_path):
from cmdforge.tool import Tool, PromptStep
@ -75,6 +90,29 @@ class TestGetStrategies:
class TestOptimizeTool:
def test_structural_results_do_not_claim_a_best_prompt(self):
from cmdforge.tool import Tool, PromptStep
tool = Tool(
name="structural", steps=[PromptStep("Summarize", "mock", "out")],
output="{out}", input_schema={"type": "string"},
output_schema={"type": "string"},
)
result = optimize_tool(tool, count=2)
assert result.best is None
assert "cannot compare prompt semantics" in result.note
def test_behavioral_evaluator_selects_only_an_improvement(self):
from cmdforge.tool import Tool, PromptStep
tool = Tool(
name="behavioral", steps=[PromptStep("Do it", "mock", "out")],
output="{out}",
)
result = optimize_tool(
tool, count=3,
evaluator=lambda candidate: len(candidate.steps[0].prompt),
)
assert result.best is not None
assert result.scores[result.best_index] > result.baseline_score
def test_baseline_score_recorded(self, tmp_path):
from cmdforge.tool import Tool, PromptStep

11
tests/test_signing.py Normal file
View File

@ -0,0 +1,11 @@
import stat
from cmdforge.signing import initialize_signing_key, load_signing_key
def test_signing_key_is_persistent_and_private(tmp_path):
path = tmp_path / "release-key.json"
created = initialize_signing_key(path)
assert load_signing_key(path) == created
assert stat.S_IMODE(path.stat().st_mode) == 0o600
assert initialize_signing_key(path) == created

67
tests/test_usage.py Normal file
View File

@ -0,0 +1,67 @@
import json
import stat
from cmdforge.usage import (
build_composite_tool,
clear_usage,
get_suggestions,
is_enabled,
record_invocation,
set_enabled,
)
class Stream:
def __init__(self, inode=None):
self.inode = inode
def configure_paths(tmp_path, monkeypatch):
monkeypatch.setattr("cmdforge.usage.CONFIG_DIR", tmp_path)
monkeypatch.setattr("cmdforge.usage.USAGE_FILE", tmp_path / "usage.json")
monkeypatch.setattr("cmdforge.usage.USAGE_LOCK_FILE", tmp_path / ".usage.lock")
def test_tracking_is_disabled_by_default(tmp_path, monkeypatch):
configure_paths(tmp_path, monkeypatch)
assert not is_enabled()
monkeypatch.setattr("cmdforge.usage._pipe_inode", lambda stream: stream.inode)
record_invocation("one", Stream(), Stream())
assert not (tmp_path / "usage.json").exists()
def test_detects_only_pipe_link_and_stores_no_content(tmp_path, monkeypatch):
configure_paths(tmp_path, monkeypatch)
set_enabled(True)
monkeypatch.setattr("cmdforge.usage._pipe_inode", lambda stream: stream.inode)
for inode in range(44, 47):
record_invocation("tool-a", Stream(), Stream(inode))
record_invocation("tool-b", Stream(inode), Stream())
assert get_suggestions() == [{
"tools": ["tool-a", "tool-b"],
"count": 3,
"last_seen": get_suggestions()[0]["last_seen"],
}]
data = json.loads((tmp_path / "usage.json").read_text())
serialized = json.dumps(data)
for forbidden in ("input_text", "output", "arguments", "cwd", "environment"):
assert forbidden not in serialized
assert stat.S_IMODE((tmp_path / "usage.json").stat().st_mode) == 0o600
def test_disable_and_clear_preserve_consent_state(tmp_path, monkeypatch):
configure_paths(tmp_path, monkeypatch)
set_enabled(True)
clear_usage()
assert is_enabled()
set_enabled(False)
assert not is_enabled()
def test_build_composite_tool_preserves_pipeline_order():
tool = build_composite_tool("combined", ["tool-a", "tool-b"])
assert tool.dependencies == ["tool-a", "tool-b"]
assert tool.steps[0].input_template == "{input}"
assert tool.steps[1].input_template == "{pipeline_1}"
assert tool.output == "{pipeline_2}"