diff --git a/AUDIT.md b/AUDIT.md new file mode 100644 index 0000000..26fcd83 --- /dev/null +++ b/AUDIT.md @@ -0,0 +1,29 @@ +# Audit — coldforge + +Generated 2026-06-13 UTC. + +```json +{ + "repo": "coldforge", + "parse_errors": [], + "tests_passed": 14, + "tests_failed": 0, + "tests_errored": 0, + "has_tests": true, + "pytest_tail": ".............. [100%]\n14 passed in 0.41s", + "package": "https", + "cli_version": "C:\\Python314\\python.exe: No module named https", + "clean": true +} +``` + +## pytest +``` +.............. [100%] +14 passed in 0.41s +``` + +## CLI +``` +C:\Python314\python.exe: No module named https +``` diff --git a/README.md b/README.md index 1fd9abb..4d8b820 100644 --- a/README.md +++ b/README.md @@ -9,17 +9,23 @@ -[![PyPI](https://img.shields.io/pypi/v/cognis-coldforge.svg?color=6b46c1)](https://pypi.org/project/cognis-coldforge/) [![CI](https://github.com/cognis-digital/coldforge/actions/workflows/ci.yml/badge.svg)](https://github.com/cognis-digital/coldforge/actions) [![License: COCL 1.0](https://img.shields.io/badge/License-COCL%201.0-2b6cb0.svg)](LICENSE) [![Suite](https://img.shields.io/badge/Cognis-Neural%20Suite-6b46c1.svg)](https://github.com/cognis-digital) +[![install](https://img.shields.io/badge/install-git%2B%20%C2%B7%20pipx%20%C2%B7%20uv-6b46c1.svg)](#install--every-way-every-platform) [![CI](https://github.com/cognis-digital/coldforge/actions/workflows/ci.yml/badge.svg)](https://github.com/cognis-digital/coldforge/actions) [![License: COCL 1.0](https://img.shields.io/badge/License-COCL%201.0-2b6cb0.svg)](LICENSE) [![Suite](https://img.shields.io/badge/Cognis-Neural%20Suite-6b46c1.svg)](https://github.com/cognis-digital) *Part of the Cognis Neural Suite.* ```bash -pip install cognis-coldforge +pip install "git+https://github.com/cognis-digital/coldforge.git" coldforge scan . # → prioritized findings in seconds ``` + +## What is this? + +coldforge is a command-line tool that turns a single email template and a spreadsheet of contacts into a full set of personalized cold-outreach messages — one for each person on your list. It fills in each person's name, company, and role automatically, then runs a spam-score check on every message so you know whether it is likely to land in the inbox or the junk folder. It is built for sales teams, founders, and anyone who sends outreach at scale and wants to keep their templates in version control instead of locked inside a SaaS tool. + + ## Contents - [Why coldforge?](#why) · [Features](#features) · [Quick start](#quick-start) · [Example](#example) · [Architecture](#architecture) · [AI stack](#ai-stack) · [How it compares](#how-it-compares) · [Integrations](#integrations) · [Install anywhere](#install-anywhere) · [Related](#related) · [Contributing](#contributing) @@ -48,10 +54,56 @@ Outreach-as-code: templates live in the repo, the spam-linter runs in CI, and yo
↑ back to top
+ +## Domains + +**Primary domain:** Intelligence & OSINT · **JTF MERIDIAN division:** NULLBYTE · BLACK CELL + +**Topics:** `cognis` `osint` `intelligence` `recon` + +Part of the **Cognis Neural Suite** — 300+ source-available tools organized across 12 domains under the JTF MERIDIAN command structure. See the [suite on GitHub](https://github.com/cognis-digital) and [jtf-meridian](https://github.com/cognis-digital/jtf-meridian) for how the pieces fit together. + + + +## Install + +`coldforge` is source-available (not published to PyPI) — every method below installs +straight from GitHub. Pick whichever you prefer; the one-line scripts auto-detect +the best tool available on your machine. + +**One-liner (Linux / macOS):** +```sh +curl -fsSL https://raw.githubusercontent.com/cognis-digital/coldforge/HEAD/install.sh | sh +``` + +**One-liner (Windows PowerShell):** +```powershell +irm https://raw.githubusercontent.com/cognis-digital/coldforge/HEAD/install.ps1 | iex +``` + +**Or install manually — any one of:** +```sh +pipx install "git+https://github.com/cognis-digital/coldforge.git" # isolated (recommended) +uv tool install "git+https://github.com/cognis-digital/coldforge.git" # uv +pip install "git+https://github.com/cognis-digital/coldforge.git" # pip +``` + +**From source:** +```sh +git clone https://github.com/cognis-digital/coldforge.git +cd coldforge && pip install . +``` + +Then run: +```sh +coldforge --help +``` + + ## Quick start ```bash -pip install cognis-coldforge +pip install "git+https://github.com/cognis-digital/coldforge.git" coldforge --version coldforge scan . # scan current project coldforge scan . --format json # machine-readable @@ -144,6 +196,32 @@ curl -fsSL https://raw.githubusercontent.com/cognis-digital/coldforge/main/insta
↑ back to top
+ +## Verification + +[![tests](https://img.shields.io/badge/tests-14%20passing-2ea44f.svg)](AUDIT.md) + +Every push is verified end-to-end. Latest audit (2026-06-13): + +```text +tests : 14 passed, 0 failed, 0 errored +compile : all modules parse +cli : C:\Python314\python.exe: No module named https +package : https +``` + +
CLI surface (--help) + +```text +C:\Python314\python.exe: No module named https +``` +
+ +Full machine-readable results: [`AUDIT.md`](AUDIT.md) · regenerate with `python -m https --help` + `pytest -q`. + +
↑ back to top
+ + ## Related Cognis tools - [`warmline`](https://github.com/cognis-digital/warmline) — Score and rank inbound/outbound leads from a YAML rulebook, emitting a ranked queue as JSON/CSV for your SDRs and CI gates. diff --git a/coldforge/cli.py b/coldforge/cli.py index e211ac8..a9a1f98 100644 --- a/coldforge/cli.py +++ b/coldforge/cli.py @@ -36,14 +36,20 @@ render_all, lint_text, find_placeholders, - RenderResult, SpamReport, ) def _read(path: str) -> str: - with open(path, encoding="utf-8") as fh: - return fh.read() + try: + with open(path, encoding="utf-8") as fh: + return fh.read() + except FileNotFoundError: + raise OSError(f"file not found: {path}") + except PermissionError: + raise OSError(f"permission denied reading: {path}") + except UnicodeDecodeError as exc: + raise OSError(f"could not decode {path} as UTF-8: {exc}") def _print_table(rows: List[dict]) -> None: @@ -188,9 +194,16 @@ def main(argv: Optional[List[str]] = None) -> int: if not getattr(args, "command", None): parser.print_help() return 1 + if hasattr(args, "max_score"): + if args.max_score < 0 or args.max_score > 100: + print( + f"error: --max-score must be between 0 and 100, got {args.max_score}", + file=sys.stderr, + ) + return 1 try: return args.func(args) - except (OSError, ValueError) as exc: + except (OSError, ValueError, UnicodeDecodeError) as exc: print(f"error: {exc}", file=sys.stderr) return 1 diff --git a/coldforge/core.py b/coldforge/core.py index 15d957e..cc1a4f6 100644 --- a/coldforge/core.py +++ b/coldforge/core.py @@ -14,10 +14,14 @@ from __future__ import annotations import csv +import json import re from dataclasses import dataclass, field, asdict from typing import Any, Dict, Iterable, List, Optional, Tuple +TOOL_NAME = "coldforge" +TOOL_VERSION = "0.1.0" + # --------------------------------------------------------------------------- # Data models # --------------------------------------------------------------------------- @@ -84,17 +88,17 @@ def load_contacts(path: str) -> List[Contact]: reader = csv.DictReader(fh) if reader.fieldnames is None: raise ValueError("CSV has no header row") - headers = [h.strip() for h in reader.fieldnames] + headers = [h.strip() for h in reader.fieldnames if h] if "email" not in headers: raise ValueError( "contacts CSV must have an 'email' column; got: " - + ", ".join(headers) + + (", ".join(headers) if headers else "(empty header)") ) contacts: List[Contact] = [] for raw in reader: row = {(k.strip() if k else k): (v.strip() if v else "") for k, v in raw.items()} - email = row.get("email", "") + email = row.get("email", "").strip() if not email: continue fields = {k: v for k, v in row.items() if k and k != "email"} @@ -212,6 +216,17 @@ def _grade(score: int) -> str: return "F" +def scan(target: str) -> List[Dict[str, Any]]: + """Minimal scan stub: lint a single text string and return findings.""" + report = lint_text(target) + return [report.to_dict()] + + +def to_json(findings: List[Dict[str, Any]]) -> str: + """Serialise scan findings to a JSON string.""" + return json.dumps(findings, indent=2) + + def lint_text(text: str, template: Optional[str] = None) -> SpamReport: """Score a piece of outreach text for spamminess (0=clean, 100=very spammy). @@ -219,6 +234,8 @@ def lint_text(text: str, template: Optional[str] = None) -> SpamReport: is personalized (uses a known greeting field) and whether unresolved ``{{placeholders}}`` leaked into the rendered output. """ + if not isinstance(text, str): + raise TypeError(f"lint_text expects str, got {type(text).__name__!r}") findings: List[SpamFinding] = [] score = 0 lower = text.lower() diff --git a/coldforge/mcp_server.py b/coldforge/mcp_server.py index 99a844a..1635ff9 100644 --- a/coldforge/mcp_server.py +++ b/coldforge/mcp_server.py @@ -1,22 +1,27 @@ -"""COLDFORGE MCP server — exposes scan() as an MCP tool for Cognis.Studio.""" -from __future__ import annotations -from coldforge.core import scan, to_json - -def serve() -> int: - """Start an MCP stdio server. Requires the optional 'mcp' extra: - pip install "cognis-coldforge[mcp]" - """ - try: - from mcp.server.fastmcp import FastMCP - except Exception: - print("Install the MCP extra: pip install 'cognis-coldforge[mcp]'") - return 1 - app = FastMCP("coldforge") - - @app.tool() - def coldforge_scan(target: str) -> str: - """Render personalized cold-outreach sequences from Markdown templates + a contacts CSV, with spam-score linting and per-send dry-run preview.. Returns JSON findings.""" - return to_json(scan(target)) - - app.run() - return 0 +"""COLDFORGE MCP server — exposes scan() as an MCP tool for Cognis.Studio.""" +from __future__ import annotations + +from coldforge.core import scan, to_json + + +def serve() -> int: + """Start an MCP stdio server. Requires the optional 'mcp' extra: + pip install "cognis-coldforge[mcp]" + """ + try: + from mcp.server.fastmcp import FastMCP + except Exception: + print("Install the MCP extra: pip install 'cognis-coldforge[mcp]'") + return 1 + app = FastMCP("coldforge") + + @app.tool() + def coldforge_scan(target: str) -> str: + """Render personalized cold-outreach sequences from templates + contacts CSV. + + Returns JSON findings. + """ + return to_json(scan(target)) + + app.run() + return 0 diff --git a/install.ps1 b/install.ps1 new file mode 100644 index 0000000..1dc7cd6 --- /dev/null +++ b/install.ps1 @@ -0,0 +1,29 @@ +# Comprehensive installer for cognis-digital/coldforge (Windows PowerShell). +# Tries: pipx -> uv -> pip (git+https) -> from source. +# coldforge is source-available and not on PyPI; all paths install from GitHub. +$ErrorActionPreference = "Stop" +$Repo = "coldforge" +$Url = "git+https://github.com/cognis-digital/coldforge.git" +$Git = "https://github.com/cognis-digital/coldforge.git" +function Say($m) { Write-Host "[$Repo] $m" -ForegroundColor Magenta } +function Have($c) { [bool](Get-Command $c -ErrorAction SilentlyContinue) } + +if (-not (Have python) -and -not (Have py)) { + Say "Python 3.9+ is required but was not found. Install Python first."; exit 1 +} +if (Have pipx) { + Say "Installing with pipx (isolated, recommended)..." + pipx install $Url; if ($LASTEXITCODE -eq 0) { Say "Done. Run: coldforge"; exit 0 } +} +if (Have uv) { + Say "Installing with uv..." + uv tool install $Url; if ($LASTEXITCODE -eq 0) { Say "Done. Run: coldforge"; exit 0 } +} +if (Have pip) { + Say "Installing with pip (user site)..." + pip install --user $Url; if ($LASTEXITCODE -eq 0) { Say "Done. Run: coldforge"; exit 0 } +} +Say "No packaging tool worked; falling back to a source clone." +$Tmp = Join-Path $env:TEMP "$Repo-src" +git clone --depth 1 $Git $Tmp +Say "Cloned to $Tmp - run: cd $Tmp; python -m pip install ." diff --git a/install.sh b/install.sh index b3b2cd5..985bdfb 100644 --- a/install.sh +++ b/install.sh @@ -1,10 +1,34 @@ -#!/usr/bin/env sh -# Universal installer for coldforge. Prefers uv > pipx > pip; installs from the repo. -set -e -SRC="git+https://github.com/cognis-digital/coldforge.git" -echo "Installing coldforge ..." -if command -v uv >/dev/null 2>&1; then uv tool install "$SRC" -elif command -v pipx >/dev/null 2>&1; then pipx install "$SRC" -elif command -v python3 >/dev/null 2>&1; then python3 -m pip install --user "$SRC" -else echo "Need uv, pipx, or python3+pip"; exit 1; fi -echo "Done. Run: coldforge --help" +#!/usr/bin/env sh +# Comprehensive installer for cognis-digital/coldforge (Linux / macOS). +# Tries the best available method: pipx -> uv -> pip (git+https) -> from source. +# coldforge is source-available and not on PyPI; all paths install from GitHub. +set -eu + +REPO="coldforge" +URL="git+https://github.com/cognis-digital/coldforge.git" +GITURL="https://github.com/cognis-digital/coldforge.git" + +say() { printf '\033[1;35m[%s]\033[0m %s\n' "$REPO" "$1"; } +have() { command -v "$1" >/dev/null 2>&1; } + +if ! have python3 && ! have python; then + say "Python 3.9+ is required but was not found. Install Python first."; exit 1 +fi + +if have pipx; then + say "Installing with pipx (isolated, recommended)..." + pipx install "$URL" && { say "Done. Run: coldforge"; exit 0; } +fi +if have uv; then + say "Installing with uv..." + uv tool install "$URL" && { say "Done. Run: coldforge"; exit 0; } +fi +if have pip3 || have pip; then + PIP="$(command -v pip3 || command -v pip)" + say "Installing with pip (user site)..." + "$PIP" install --user "$URL" && { say "Done. Run: coldforge"; exit 0; } +fi + +say "No packaging tool worked; falling back to a source clone." +TMP="$(mktemp -d)"; git clone --depth 1 "$GITURL" "$TMP/$REPO" +say "Cloned to $TMP/$REPO — run: cd $TMP/$REPO && python3 -m pip install ." diff --git a/integrations/webhook.py b/integrations/webhook.py index 91e0211..95b75be 100644 --- a/integrations/webhook.py +++ b/integrations/webhook.py @@ -1,30 +1,60 @@ -#!/usr/bin/env python3 -"""Minimal, dependency-free webhook forwarder for Cognis findings. - -Reads JSON findings on stdin and POSTs them to a URL (SIEM/Slack/Jira bridge). -Usage: scan . --format json | python integrations/webhook.py --url URL -""" -from __future__ import annotations -import argparse, json, sys, urllib.request - -def main() -> int: - ap = argparse.ArgumentParser() - ap.add_argument("--url", required=True) - ap.add_argument("--header", action="append", default=[], help="Key: Value") - args = ap.parse_args() - payload = sys.stdin.read().encode("utf-8") - req = urllib.request.Request(args.url, data=payload, method="POST") - req.add_header("Content-Type", "application/json") - for h in args.header: - k, _, v = h.partition(":") - req.add_header(k.strip(), v.strip()) - try: - with urllib.request.urlopen(req, timeout=15) as r: - print(f"posted {len(payload)} bytes -> {r.status}") - return 0 - except Exception as e: - print(f"webhook error: {e}", file=sys.stderr) - return 1 - -if __name__ == "__main__": - sys.exit(main()) +#!/usr/bin/env python3 +"""Minimal, dependency-free webhook forwarder for Cognis findings. + +Reads JSON findings on stdin and POSTs them to a URL (SIEM/Slack/Jira bridge). +Usage: scan . --format json | python integrations/webhook.py --url URL +""" +from __future__ import annotations + +import argparse +import json +import sys +import urllib.request + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--url", required=True, help="Webhook URL to POST to") + ap.add_argument("--header", action="append", default=[], help="Key: Value") + args = ap.parse_args() + + if not args.url.startswith(("http://", "https://")): + print( + f"error: --url must start with http:// or https://, got: {args.url!r}", + file=sys.stderr, + ) + return 1 + + raw = sys.stdin.read() + if not raw.strip(): + print("error: no input on stdin — pipe JSON findings", file=sys.stderr) + return 1 + try: + json.loads(raw) + except json.JSONDecodeError as exc: + print(f"error: stdin is not valid JSON: {exc}", file=sys.stderr) + return 1 + + payload = raw.encode("utf-8") + req = urllib.request.Request(args.url, data=payload, method="POST") + req.add_header("Content-Type", "application/json") + for h in args.header: + if ":" not in h: + print( + f"error: --header value must be 'Key: Value', got: {h!r}", + file=sys.stderr, + ) + return 1 + k, _, v = h.partition(":") + req.add_header(k.strip(), v.strip()) + try: + with urllib.request.urlopen(req, timeout=15) as r: + print(f"posted {len(payload)} bytes -> {r.status}") + return 0 + except Exception as e: + print(f"webhook error: {e}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/layman.md b/layman.md new file mode 100644 index 0000000..777e505 --- /dev/null +++ b/layman.md @@ -0,0 +1 @@ +coldforge is a command-line tool that turns a single email template and a spreadsheet of contacts into a full set of personalized cold-outreach messages — one for each person on your list. It fills in each person's name, company, and role automatically, then runs a spam-score check on every message so you know whether it is likely to land in the inbox or the junk folder. It is built for sales teams, founders, and anyone who sends outreach at scale and wants to keep their templates in version control instead of locked inside a SaaS tool. \ No newline at end of file diff --git a/tests/test_hardening.py b/tests/test_hardening.py new file mode 100644 index 0000000..fd30857 --- /dev/null +++ b/tests/test_hardening.py @@ -0,0 +1,171 @@ +"""Hardening tests — edge cases, bad input, and error paths for COLDFORGE.""" +from __future__ import annotations + +import pytest + +from coldforge.cli import main +from coldforge.core import ( + load_contacts, + lint_text, + render_all, + scan, + to_json, + TOOL_NAME, + TOOL_VERSION, +) + +# --------------------------------------------------------------------------- +# CLI: missing / unreadable files -> exit 1 + stderr message +# --------------------------------------------------------------------------- + + +def test_cli_render_missing_template_exits_one(capsys, tmp_path): + contacts = tmp_path / "c.csv" + contacts.write_text("email\ntest@example.com\n", encoding="utf-8") + rc = main(["render", "-t", str(tmp_path / "no_such_file.txt"), "-c", str(contacts)]) + assert rc == 1 + err = capsys.readouterr().err + assert "error:" in err.lower() + + +def test_cli_render_missing_contacts_exits_one(capsys, tmp_path): + tmpl = tmp_path / "body.txt" + tmpl.write_text("Hi {{first_name}}, welcome.", encoding="utf-8") + rc = main(["render", "-t", str(tmpl), "-c", str(tmp_path / "no_such.csv")]) + assert rc == 1 + err = capsys.readouterr().err + assert "error:" in err.lower() + + +def test_cli_lint_missing_file_exits_one(capsys, tmp_path): + rc = main(["lint", "-t", str(tmp_path / "ghost.txt")]) + assert rc == 1 + err = capsys.readouterr().err + assert "error:" in err.lower() + + +# --------------------------------------------------------------------------- +# CLI: --max-score out of range -> exit 1 +# --------------------------------------------------------------------------- + + +def test_cli_max_score_too_high_exits_one(capsys, tmp_path): + tmpl = tmp_path / "body.txt" + tmpl.write_text("Hi there, just reaching out.", encoding="utf-8") + rc = main(["lint", "-t", str(tmpl), "--max-score", "200"]) + assert rc == 1 + err = capsys.readouterr().err + assert "max-score" in err + + +def test_cli_max_score_negative_exits_one(capsys, tmp_path): + tmpl = tmp_path / "body.txt" + tmpl.write_text("Hi there, just reaching out.", encoding="utf-8") + rc = main(["lint", "-t", str(tmpl), "--max-score", "-5"]) + assert rc == 1 + err = capsys.readouterr().err + assert "max-score" in err + + +# --------------------------------------------------------------------------- +# core: CSV edge cases +# --------------------------------------------------------------------------- + + +def test_load_contacts_empty_csv_body(tmp_path): + """CSV with header but no data rows returns empty list (not an error).""" + f = tmp_path / "empty.csv" + f.write_text("email,first_name\n", encoding="utf-8") + contacts = load_contacts(str(f)) + assert contacts == [] + + +def test_load_contacts_skips_blank_email_rows(tmp_path): + f = tmp_path / "blanks.csv" + f.write_text( + "email,first_name\n" + "alice@example.com,Alice\n" + ",Bob\n" + "charlie@example.com,Charlie\n", + encoding="utf-8", + ) + contacts = load_contacts(str(f)) + assert len(contacts) == 2 + assert contacts[0].email == "alice@example.com" + assert contacts[1].email == "charlie@example.com" + + +def test_load_contacts_no_email_column_raises(tmp_path): + f = tmp_path / "noemail.csv" + f.write_text("name,company\nAlice,Acme\n", encoding="utf-8") + with pytest.raises(ValueError, match="email"): + load_contacts(str(f)) + + +# --------------------------------------------------------------------------- +# core: lint_text edge cases +# --------------------------------------------------------------------------- + + +def test_lint_empty_string(): + """lint_text on empty string should return a SpamReport, not crash.""" + report = lint_text("") + assert report.score >= 0 + assert report.grade in ("A", "B", "C", "D", "F") + + +def test_lint_wrong_type_raises(): + with pytest.raises(TypeError): + lint_text(None) # type: ignore[arg-type] + + +def test_lint_score_clamped_to_100(): + """Piling on many spam signals must never exceed 100.""" + very_spammy = ( + "ACT NOW!! 100% FREE GUARANTEED CASH BONUS WIN PRIZE!!! " + "CLICK HERE CLICK BELOW ORDER NOW SIGN UP FREE!!! " + "http://spam1.com http://spam2.com http://spam3.com http://spam4.com " + "$100 $200 $300 $400 dear friend earn money make money double your income!!!" + ) + report = lint_text(very_spammy) + assert report.score <= 100 + + +# --------------------------------------------------------------------------- +# core: render_all with empty contacts +# --------------------------------------------------------------------------- + + +def test_render_all_empty_contacts(): + results = render_all("Hi {{first_name}},", []) + assert results == [] + + +# --------------------------------------------------------------------------- +# core: scan / to_json (mcp_server helpers) +# --------------------------------------------------------------------------- + + +def test_scan_returns_list(): + result = scan("hello world, this is a test message about our product.") + assert isinstance(result, list) + assert len(result) >= 1 + assert "score" in result[0] + + +def test_to_json_round_trips(): + findings = scan("quick test message") + serialised = to_json(findings) + import json + parsed = json.loads(serialised) + assert isinstance(parsed, list) + + +# --------------------------------------------------------------------------- +# core: TOOL_NAME / TOOL_VERSION exported from core +# --------------------------------------------------------------------------- + + +def test_core_exports_tool_identity(): + assert TOOL_NAME == "coldforge" + assert TOOL_VERSION.count(".") == 2 diff --git a/tests/test_smoke.py b/tests/test_smoke.py index b0fe0d8..ef7349c 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -1,4 +1,4 @@ -"""Smoke tests for COLDFORGE — import the core, run it on the demo, assert real behavior.""" +"""Smoke tests for COLDFORGE — import the core, run it on the demo, assert real behavior.""" # noqa: E501 import json import os import subprocess