📦 deps(thirdparty): update snapshots

This commit is contained in:
ci[bot]
2026-07-03 16:04:10 +00:00
parent 2bf579321a
commit b40381458b
482 changed files with 8324 additions and 2007 deletions
@@ -27,6 +27,19 @@ import time
from datetime import datetime, timezone
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
# ---------------------------------------------------------------------------
# Imports from the 007 config hub (same directory)
# ---------------------------------------------------------------------------
@@ -397,31 +410,30 @@ def _phase1_surface_mapping(target: Path, verbose: bool = False) -> dict:
_config_extensions = {".json", ".yaml", ".yml", ".toml", ".ini", ".cfg", ".conf", ".env"}
for root, dirs, filenames in os.walk(target):
dirs[:] = [d for d in dirs if d not in SKIP_DIRECTORIES]
for fpath in safe_user_path(target).rglob("*"):
if not fpath.is_file() or any(part in SKIP_DIRECTORIES for part in fpath.parts):
continue
fname = fpath.name
total_files += 1
suffix = fpath.suffix.lower()
for fname in filenames:
total_files += 1
fpath = Path(root) / fname
suffix = fpath.suffix.lower()
# Categorize by extension
ext_key = suffix if suffix else "(no extension)"
files_by_type[ext_key] = files_by_type.get(ext_key, 0) + 1
# Categorize by extension
ext_key = suffix if suffix else "(no extension)"
files_by_type[ext_key] = files_by_type.get(ext_key, 0) + 1
# Detect entry points
for pat in _entry_point_patterns:
if pat.search(fname) or pat.search(str(fpath)):
entry_points.append(str(fpath))
break
# Detect entry points
for pat in _entry_point_patterns:
if pat.search(fname) or pat.search(str(fpath)):
entry_points.append(str(fpath))
break
# Detect dependency files
if fname.lower() in _dep_file_names:
dependency_files.append(str(fpath))
# Detect dependency files
if fname.lower() in _dep_file_names:
dependency_files.append(str(fpath))
# Detect config files
if suffix in _config_extensions or fname.lower().startswith(".env"):
config_files.append(str(fpath))
# Detect config files
if suffix in _config_extensions or fname.lower().startswith(".env"):
config_files.append(str(fpath))
# Sort by count descending
sorted_types = sorted(files_by_type.items(), key=lambda x: x[1], reverse=True)
@@ -1053,7 +1065,7 @@ def run_audit(
ensure_directories()
target = Path(target_path).resolve()
target = safe_user_path(target_path).resolve()
if not target.exists():
logger.error("Target path does not exist: %s", target)
sys.exit(1)
@@ -17,6 +17,19 @@ import sys
import time
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
# ---------------------------------------------------------------------------
# Imports from the 007 config hub (same directory)
# ---------------------------------------------------------------------------
@@ -134,20 +147,16 @@ def collect_files(target: Path, logger) -> list[Path]:
files: list[Path] = []
max_files = LIMITS["max_files_per_scan"]
for root, dirs, filenames in os.walk(target):
# Prune skipped directories in-place so os.walk does not descend
dirs[:] = [d for d in dirs if not _should_skip_dir(d)]
for fname in filenames:
if len(files) >= max_files:
logger.warning(
"Reached max_files_per_scan limit (%d). Stopping collection.", max_files
)
return files
fpath = Path(root) / fname
if _is_scannable(fpath):
files.append(fpath)
for fpath in safe_user_path(target).rglob("*"):
if not fpath.is_file() or any(_should_skip_dir(part) for part in fpath.parts):
continue
if len(files) >= max_files:
logger.warning(
"Reached max_files_per_scan limit (%d). Stopping collection.", max_files
)
return files
if _is_scannable(fpath):
files.append(fpath)
return files
@@ -368,7 +377,7 @@ def run_scan(target_path: str, output_format: str = "text", verbose: bool = Fals
logger = setup_logging("007-quick-scan")
ensure_directories()
target = Path(target_path).resolve()
target = safe_user_path(target_path).resolve()
if not target.exists():
logger.error("Target path does not exist: %s", target)
sys.exit(1)
@@ -17,6 +17,19 @@ import sys
import time
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
# ---------------------------------------------------------------------------
# Import from the 007 config hub (parent directory)
# ---------------------------------------------------------------------------
@@ -850,27 +863,26 @@ def discover_dependency_files(target: Path) -> list[Path]:
"""
found: list[Path] = []
for root, dirs, filenames in os.walk(target):
dirs[:] = [d for d in dirs if d not in config.SKIP_DIRECTORIES]
for fpath in safe_user_path(target).rglob("*"):
if not fpath.is_file() or any(part in config.SKIP_DIRECTORIES for part in fpath.parts):
continue
fname = fpath.name
fname_lower = fname.lower()
for fname in filenames:
fpath = Path(root) / fname
fname_lower = fname.lower()
# Exact name matches
if fname in ALL_DEP_FILES:
found.append(fpath)
continue
# Exact name matches
if fname in ALL_DEP_FILES:
found.append(fpath)
continue
# requirements*.txt variants
if _REQUIREMENTS_RE.match(fname):
found.append(fpath)
continue
# requirements*.txt variants
if _REQUIREMENTS_RE.match(fname):
found.append(fpath)
continue
# Docker files (prefix match)
if any(fname_lower.startswith(prefix.lower()) for prefix in DOCKER_PREFIXES):
found.append(fpath)
continue
# Docker files (prefix match)
if any(fname_lower.startswith(prefix.lower()) for prefix in DOCKER_PREFIXES):
found.append(fpath)
continue
return found
@@ -1158,7 +1170,7 @@ def run_scan(
config.ensure_directories()
target = Path(target_path).resolve()
target = safe_user_path(target_path).resolve()
if not target.exists():
logger.error("Target path does not exist: %s", target)
sys.exit(1)
@@ -20,6 +20,19 @@ import sys
import time
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
# ---------------------------------------------------------------------------
# Import from the 007 config hub (parent directory)
# ---------------------------------------------------------------------------
@@ -546,19 +559,16 @@ def collect_files(target: Path) -> list[Path]:
files: list[Path] = []
max_files = config.LIMITS["max_files_per_scan"]
for root, dirs, filenames in os.walk(target):
dirs[:] = [d for d in dirs if d not in config.SKIP_DIRECTORIES]
for fname in filenames:
if len(files) >= max_files:
logger.warning(
"Reached max_files_per_scan limit (%d). Stopping.", max_files
)
return files
fpath = Path(root) / fname
if _should_scan_file(fpath):
files.append(fpath)
for fpath in safe_user_path(target).rglob("*"):
if not fpath.is_file() or any(part in config.SKIP_DIRECTORIES for part in fpath.parts):
continue
if len(files) >= max_files:
logger.warning(
"Reached max_files_per_scan limit (%d). Stopping.", max_files
)
return files
if _should_scan_file(fpath):
files.append(fpath)
return files
@@ -961,7 +971,7 @@ def run_scan(
config.ensure_directories()
target = Path(target_path).resolve()
target = safe_user_path(target_path).resolve()
if not target.exists():
logger.error("Target path does not exist: %s", target)
sys.exit(1)
@@ -20,6 +20,19 @@ import sys
import time
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
# ---------------------------------------------------------------------------
# Import from the 007 config hub (parent directory)
# ---------------------------------------------------------------------------
@@ -375,19 +388,16 @@ def collect_files(target: Path) -> list[Path]:
files: list[Path] = []
max_files = config.LIMITS["max_files_per_scan"]
for root, dirs, filenames in os.walk(target):
dirs[:] = [d for d in dirs if d not in config.SKIP_DIRECTORIES]
for fname in filenames:
if len(files) >= max_files:
logger.warning(
"Reached max_files_per_scan limit (%d). Stopping.", max_files
)
return files
fpath = Path(root) / fname
if _should_scan_file(fpath):
files.append(fpath)
for fpath in safe_user_path(target).rglob("*"):
if not fpath.is_file() or any(part in config.SKIP_DIRECTORIES for part in fpath.parts):
continue
if len(files) >= max_files:
logger.warning(
"Reached max_files_per_scan limit (%d). Stopping.", max_files
)
return files
if _should_scan_file(fpath):
files.append(fpath)
return files
@@ -869,7 +879,7 @@ def run_scan(
config.ensure_directories()
target = Path(target_path).resolve()
target = safe_user_path(target_path).resolve()
if not target.exists():
logger.error("Target path does not exist: %s", target)
sys.exit(1)
@@ -24,6 +24,19 @@ import sys
import time
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
# ---------------------------------------------------------------------------
# Imports from the 007 config hub (same directory)
# ---------------------------------------------------------------------------
@@ -141,18 +154,17 @@ def _collect_source_files(target: Path) -> list[Path]:
files: list[Path] = []
max_files = LIMITS["max_files_per_scan"]
for root, dirs, filenames in os.walk(target):
dirs[:] = [d for d in dirs if d not in SKIP_DIRECTORIES]
for fname in filenames:
if len(files) >= max_files:
return files
fpath = Path(root) / fname
suffix = fpath.suffix.lower()
name = fpath.name.lower()
for ext in SCANNABLE_EXTENSIONS:
if name.endswith(ext) or suffix == ext:
files.append(fpath)
break
for fpath in safe_user_path(target).rglob("*"):
if not fpath.is_file() or any(part in SKIP_DIRECTORIES for part in fpath.parts):
continue
if len(files) >= max_files:
return files
suffix = fpath.suffix.lower()
name = fpath.name.lower()
for ext in SCANNABLE_EXTENSIONS:
if name.endswith(ext) or suffix == ext:
files.append(fpath)
break
return files
@@ -529,7 +541,7 @@ def run_score(
ensure_directories()
target = Path(target_path).resolve()
target = safe_user_path(target_path).resolve()
if not target.exists():
logger.error("Target path does not exist: %s", target)
sys.exit(1)
@@ -14,6 +14,20 @@ import socket
import requests
from urllib.parse import urlparse
from typing import Optional, Dict, Any
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
API_BASE_URL = "https://2slides.com/api/v1"
@@ -124,7 +138,7 @@ def download_slides_pages_voices(
zip_response.raise_for_status()
# Save to file
with open(output_path, 'wb') as f:
with safe_user_path(output_path).open('wb') as f:
for chunk in zip_response.iter_content(chunk_size=8192):
f.write(chunk)
@@ -12,6 +12,19 @@ import shutil
from datetime import datetime
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
# Rich for beautiful terminal output
try:
from rich.console import Console
@@ -386,9 +399,9 @@ def save_outputs(transcript_text, ata_text, audio_file, output_dir="."):
# Sempre salva transcript
transcript_filename = f"transcript-{timestamp}.md"
transcript_path = Path(output_dir) / transcript_filename
transcript_path = safe_user_path(output_dir) / transcript_filename
with open(transcript_path, 'w', encoding='utf-8') as f:
with transcript_path.open('w', encoding='utf-8') as f:
f.write(transcript_text)
console.print(f"[green]✅ Transcript salvo:[/green] {transcript_filename}")
@@ -397,9 +410,9 @@ def save_outputs(transcript_text, ata_text, audio_file, output_dir="."):
ata_path = None
if ata_text:
ata_filename = f"ata-{timestamp}.md"
ata_path = Path(output_dir) / ata_filename
ata_path = safe_user_path(output_dir) / ata_filename
with open(ata_path, 'w', encoding='utf-8') as f:
with ata_path.open('w', encoding='utf-8') as f:
f.write(ata_text)
console.print(f"[green]✅ Ata salva:[/green] {ata_filename}")
@@ -30,6 +30,19 @@ import sys
from collections import Counter, defaultdict
from datetime import datetime, timezone
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from typing import Any
@@ -336,11 +349,11 @@ def build_report(records: list[dict[str, Any]], source_path: Path) -> str:
def main(argv: list[str]) -> int:
if len(argv) != 3:
print(f"usage: {Path(argv[0]).name} <input.ndjson> <output.md>", file=sys.stderr)
print(f"usage: {safe_user_path(argv[0]).name} <input.ndjson> <output.md>", file=sys.stderr)
return 1
input_path = Path(argv[1])
output_path = Path(argv[2])
input_path = safe_user_path(argv[1])
output_path = safe_user_path(argv[2])
if not input_path.exists() or input_path.stat().st_size == 0:
print(f"error: {input_path} is missing or empty", file=sys.stderr)
@@ -20,6 +20,19 @@ import time
from datetime import datetime
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
try:
import psutil
except ImportError:
@@ -281,8 +294,8 @@ def main():
else:
output_path = f"monitor_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(output_path, "w", encoding="utf-8") as f:
json.dump(output_data, f, indent=2, ensure_ascii=False)
with safe_user_path(output_path).open("w", encoding="utf-8") as f:
f.write(json.dumps(output_data, indent=2, ensure_ascii=False))
print(f"\nLog salvo em: {output_path}\n")
@@ -0,0 +1,192 @@
---
name: code-polish
description: Rewrites unprofessional code comments into clear ones and performs non-semantic cleanup. Use to professionalize code without altering logic or behavior.
risk: critical
source: community
date_added: "2026-07-02"
---
# Code Polish
A constraint-based protocol for normalizing code comments and performing safe, non-semantic cleanup. This skill exists because human-written code tends to carry casual, outdated, or missing comments, while the goal is professional-grade documentation without touching behavior.
This file is self-contained. Do not require any other skill file to execute this protocol.
## Prime Directive
Comments and non-semantic cleanup are the job. Logic is never the job. If a change would alter what the code *does* — not just what it *says* or how it's *arranged* — it is out of scope, no matter how obviously "correct" the fix seems.
---
## When to Use
Apply this skill when:
- The user asks to "clean up," "professionalize," or "polish" existing code
- Code is being prepped for code review, handoff, open-sourcing, or documentation
- A file has a mix of human and AI-written comments and needs one consistent, professional voice
- Comments are outdated, missing, redundant, or written casually (venting, placeholders, inside jokes)
- The user wants comments improved but explicitly does **not** want logic touched
Do not apply this skill when:
- The user wants a bug fixed or behavior changed (that's a different job — logic edits are out of scope here)
- The user wants a full rewrite or architectural restructuring
- The only ask is adding new features or functionality
---
## Phase 0 — Full Read
Before editing anything, read the entire file (or the entire relevant module if the codebase is large — not just the function in question). Do not comment or clean incrementally while still reading. A comment written without full context is a guess, and guesses are how "professional" comments end up wrong.
Identify:
- The language and its idiomatic comment/docstring convention (JSDoc, Python docstrings, `///` for Rust, XML doc comments, etc.)
- Any existing project comment style already in use elsewhere in the file — match it rather than importing a foreign convention
- Any comment that encodes real, non-obvious information (race conditions, workarounds for external bugs, "don't reorder this" warnings, business-rule justifications)
---
## Phase 1 — Comment Audit
Classify every existing comment into one of these categories before touching it:
| Category | Example | Action |
|---|---|---|
| **Junk / venting** | `// wtf is this`, `// idk why but it works` | Remove tone, extract any real information underneath, rewrite professionally — or delete if it truly holds zero information |
| **Placeholder** | `// fix later`, `// TODO hack` | Convert to a proper `TODO:` note with the actual concern stated plainly, or remove if stale/resolved |
| **Dead code comments** | Blocks of commented-out code | Remove, unless the surrounding context makes clear it's intentionally preserved (e.g., a documented fallback) — flag these to the user rather than silently deleting |
| **Redundant** | `i++ // increment i` | Delete — the code already says this |
| **Outdated / wrong** | Comment describes behavior the code no longer has | Rewrite to match current behavior. Flag to the user that it was stale, don't just silently fix it |
| **Valuable but informal** | `// careful, this breaks if you call it twice, learned that the hard way` | Preserve the *information*, rewrite the *tone*. Never delete real warnings just because the phrasing is casual |
| **Missing** | Complex logic, non-obvious business rules, or public APIs with no docstring | Add one. Don't over-comment simple, self-explanatory lines |
---
## Phase 2 — Non-Semantic Cleanup
Scope is strictly limited to changes that cannot alter behavior:
- Consistent indentation and whitespace
- Consistent brace/bracket style matching the surrounding file
- Removing truly dead code (unreachable blocks) — only when unambiguous, and flagged in the summary
- Splitting overly long lines for readability
- Local variable renaming for clarity is allowed **only** for private/local-scope names, and only when the improvement is unambiguous — never rename anything exported, public, or referenced across files without calling it out explicitly first
Anything beyond this — reordering logic, extracting functions, changing control flow, altering algorithms — is out of scope for this skill.
---
## Phase 3 — Comment Rewrite / Addition
Apply these standards to every comment touched or added:
- **Explain why, not what.** The code already shows *what* it does; a comment earns its place by explaining intent, tradeoffs, or non-obvious constraints.
- **Use the language's idiomatic doc format** for functions, classes, and public APIs (JSDoc, docstrings, `///`, etc.) — match the convention already used elsewhere in the file if one exists.
- **Be concise.** No padding, no restating the obvious, no filler sentences.
- **No informal register.** No jokes, no venting, no first-person asides ("I think this works because...").
- **No AI-tell phrasing.** Avoid generic filler like "This function is responsible for..." or "Note that..." padding, and avoid em-dashes. Write plainly and directly, the way a careful senior engineer would.
- **Don't invent behavior.** If you're not certain why something is done a certain way, say what the code does, not a fabricated justification for why.
---
## Phase 4 — Verification
Before presenting the result:
- Confirm the edited file's logic is behaviorally identical to the original — comments and whitespace are the only permitted diffs, plus whatever narrow Phase 2 cleanup was done.
- Re-read the diff end to end, not just the changed lines in isolation, to catch anything that accidentally shifted meaning.
- If a rewritten comment removes information that was present in the original (even informally stated), that's a failure — go back and preserve it.
---
## Phase 5 — Report Back
Summarize for the user, don't just hand back a silent diff:
- How many comments were rewritten, added, or removed, and why
- Any comments flagged as "informal but contained a real warning" — confirm the information was preserved
- Any dead code or stale comments removed, listed explicitly
- Anything you were unsure about and left alone rather than guessing
---
## Examples
**Junk / venting → professional**
```js
// before
// ugh this took forever to figure out. api rate limits us super hard in prod so we have to do exponential backoff here. just leave it alone
function retryFetch(url, attempts) { ... }
// after
// Uses exponential backoff to handle aggressive API rate-limiting in production.
function retryFetch(url, attempts) { ... }
```
**Redundant → removed**
```python
# before
count += 1 # increment count by 1
# after
count += 1
```
**Valuable but informal → tone rewritten, information preserved**
```python
# before
# careful, this breaks if you call it twice, learned that the hard way
# after
# Not idempotent: calling this more than once per session corrupts the
# cache index. Callers must guard against duplicate invocation.
```
**Missing → added**
```java
// before
public double calculate(double base, int tier) {
return base * (tier > 2 ? 0.85 : 1.0);
}
// after
/**
* Applies the loyalty discount. Tiers above 2 qualify for a 15% discount;
* this threshold matches the current pricing policy, not a technical limit.
*/
public double calculate(double base, int tier) {
return base * (tier > 2 ? 0.85 : 1.0);
}
```
**Outdated / wrong → corrected and flagged**
```go
// before
// returns nil if user not found
func GetUser(id string) (*User, error) { ... } // now returns ErrNotFound instead
// after
// Returns ErrNotFound if the user does not exist.
func GetUser(id string) (*User, error) { ... }
// (flagged to user: original comment was stale — function used to return nil,
// now returns a named error)
```
---
## Security & Safety Notes
This skill never:
- Changes program logic, control flow, or algorithmic behavior
- Restructures code (extracting/inlining functions, reordering execution, changing architecture)
- Renames anything public, exported, or cross-referenced without explicit confirmation
- Deletes a comment solely because its tone is casual, without checking whether it carries real information first
- Fabricates a rationale for a comment when the actual reason isn't knowable from context — state what's certain only
---
## Limitations
- Cannot verify runtime behavior — Phase 4 is a read-through diff check, not a test run. For anything beyond trivial files, the user should still run the actual test suite after applying this skill.
- Judgment calls on ambiguous cases (e.g., "is this dead code intentional or forgotten?") default to flagging rather than guessing — this means some cleanup will need a quick human yes/no rather than happening silently.
- Not a substitute for a linter or formatter — Phase 2 cleanup is deliberately conservative and won't enforce a full style guide (e.g., max line length rules, import ordering) unless that's trivially inferable from the surrounding file.
- Comment quality is bounded by how well the code's actual intent can be inferred from context. If the "why" genuinely isn't recoverable from the file (no domain knowledge, no commit history, no ticket references available), the honest output is a comment describing *what*, not a confident but invented *why*.
- Large files or unfamiliar codebases increase the risk of Phase 0 missing context that would have changed a comment's wording — flag uncertainty in the Phase 5 report rather than presenting low-confidence rewrites as settled.
@@ -12,12 +12,33 @@
//
// Usage: node capture_screenshots.mjs <research-dir> [--mode remote|local] [--concurrency 2]
import sanitizeFilename from 'sanitize-filename';
import { readdirSync, readFileSync, mkdirSync, existsSync } from 'fs';
import { join } from 'path';
import { isAbsolute, join, relative, resolve } from 'path';
import { spawnSync } from 'child_process';
import { parseFrontmatter } from './md_utils.mjs';
const args = process.argv.slice(2);
function sanitizePathSegments(pathValue) {
return String(pathValue ?? '').split(/[\\/]+/).filter(Boolean).map((segment) => {
const sanitized = sanitizeFilename(segment);
if (sanitized !== segment || !sanitized) {
throw new Error(`Unsafe path segment: ${segment}`);
}
return sanitized;
});
}
function safeCliPath(pathValue, baseDir = process.cwd()) {
const root = resolve(baseDir);
const target = resolve(root, ...sanitizePathSegments(pathValue));
const rel = relative(root, target);
if (rel.startsWith('..') || isAbsolute(rel)) {
throw new Error(`Path escapes allowed directory: ${pathValue}`);
}
return target;
}
if (args.includes('--help') || args.includes('-h') || args.length === 0) {
console.error(`Usage: node capture_screenshots.mjs <research-dir> [options]
@@ -38,7 +59,7 @@ Options:
process.exit(args.includes('--help') || args.includes('-h') ? 0 : 1);
}
const dir = args[0];
const dir = safeCliPath(args[0]);
const modeIdx = args.indexOf('--mode');
const browseMode = modeIdx !== -1 ? args[modeIdx + 1] : 'remote';
const modeFlag = browseMode === 'local' ? '--local' : '--remote';
@@ -63,7 +84,7 @@ if (concurrency > 1) {
concurrency = 1;
}
const shotsDir = join(dir, 'screenshots');
const shotsDir = safeCliPath('screenshots', dir);
mkdirSync(shotsDir, { recursive: true });
function run(cmd, args, { timeout = 30000 } = {}) {
@@ -9,6 +9,7 @@
import { readdirSync, readFileSync, writeFileSync, existsSync, mkdirSync } from 'fs';
import { basename, dirname, join, relative, resolve } from 'path';
import { fileURLToPath } from 'url';
import sanitizeFilename from 'sanitize-filename';
import { parseFrontmatter, parseBody, parseSections } from './md_utils.mjs';
const __filename = fileURLToPath(import.meta.url);
@@ -17,9 +18,19 @@ const __dirname = dirname(__filename);
const args = process.argv.slice(2);
const SAFE_SLUG_RE = /^[A-Za-z0-9][A-Za-z0-9._-]*$/;
function sanitizePathSegments(pathValue) {
return String(pathValue ?? '').split(/[\\/]+/).filter(Boolean).map((segment) => {
const sanitized = sanitizeFilename(segment);
if (sanitized !== segment || !sanitized) {
throw new Error(`Unsafe path segment: ${segment}`);
}
return sanitized;
});
}
function safeJoin(base, ...parts) {
const root = resolve(base);
const target = resolve(root, ...parts);
const target = resolve(root, ...parts.flatMap(sanitizePathSegments));
const rel = relative(root, target);
if (rel.startsWith('..') || rel.startsWith('/')) {
throw new Error(`Path escapes research directory: ${parts.join('/')}`);
@@ -32,7 +43,7 @@ function safeResearchDir(rawDir) {
throw new Error('Research directory is required');
}
const root = resolve(process.cwd());
const target = resolve(root, rawDir);
const target = safeJoin(root, rawDir);
const rel = relative(root, target);
if ((rel.startsWith('..') || rel.startsWith('/')) && process.env.COMPETITOR_ANALYSIS_ALLOW_EXTERNAL_DIR !== '1') {
throw new Error('Research directory must stay under the current working directory');
@@ -9,10 +9,31 @@
// Output: newline-delimited JSON to stdout, one object per candidate:
// { "name": "serper", "hits": 3, "domain": "serper.dev", "example": "Tavily vs Serper..." }
import sanitizeFilename from 'sanitize-filename';
import { readdirSync, readFileSync } from 'fs';
import { join } from 'path';
import { isAbsolute, join, relative, resolve } from 'path';
const args = process.argv.slice(2);
function sanitizePathSegments(pathValue) {
return String(pathValue ?? '').split(/[\\/]+/).filter(Boolean).map((segment) => {
const sanitized = sanitizeFilename(segment);
if (sanitized !== segment || !sanitized) {
throw new Error(`Unsafe path segment: ${segment}`);
}
return sanitized;
});
}
function safeCliPath(pathValue, baseDir = process.cwd()) {
const root = resolve(baseDir);
const target = resolve(root, ...sanitizePathSegments(pathValue));
const rel = relative(root, target);
if (rel.startsWith('..') || isAbsolute(rel)) {
throw new Error(`Path escapes allowed directory: ${pathValue}`);
}
return target;
}
if (args.includes('--help') || args.includes('-h') || args.length === 0) {
console.error(`Usage: node extract_vs_names.mjs <directory> [--prefix <prefix>] [--seed "<csv>"]
@@ -28,7 +49,7 @@ Options:
process.exit(args.includes('--help') || args.includes('-h') ? 0 : 1);
}
const dir = args[0];
const dir = safeCliPath(args[0]);
const prefixIdx = args.indexOf('--prefix');
const prefix = prefixIdx !== -1 && args[prefixIdx + 1] ? args[prefixIdx + 1] : 'competitor';
const seedIdx = args.indexOf('--seed');
@@ -14,6 +14,7 @@
// { "url": "https://foo.com", "status": "PASS" | "REJECT" | "UNKNOWN",
// "matched_includes": [...], "matched_excludes": [...], "title": "...", "hero": "..." }
import sanitizeFilename from 'sanitize-filename';
import { execFile } from 'child_process';
import { promisify } from 'util';
import { readFileSync } from 'fs';
@@ -25,6 +26,26 @@ import { readFileSync } from 'fs';
const execFileAsync = promisify(execFile);
const args = process.argv.slice(2);
function sanitizePathSegments(pathValue) {
return String(pathValue ?? '').split(/[\\/]+/).filter(Boolean).map((segment) => {
const sanitized = sanitizeFilename(segment);
if (sanitized !== segment || !sanitized) {
throw new Error(`Unsafe path segment: ${segment}`);
}
return sanitized;
});
}
function safeCliPath(pathValue, baseDir = process.cwd()) {
const root = resolve(baseDir);
const target = resolve(root, ...sanitizePathSegments(pathValue));
const rel = relative(root, target);
if (rel.startsWith('..') || isAbsolute(rel)) {
throw new Error(`Path escapes allowed directory: ${pathValue}`);
}
return target;
}
if (args.includes('--help') || args.includes('-h')) {
console.error(`Usage: cat urls.txt | node gate_candidates.mjs [options]
@@ -5,10 +5,31 @@
// Reads all {prefix}_discovery_batch_*.json files, deduplicates by domain,
// outputs one URL per line to stdout, stats to stderr.
import sanitizeFilename from 'sanitize-filename';
import { readdirSync, readFileSync } from 'fs';
import { join } from 'path';
import { isAbsolute, join, relative, resolve } from 'path';
const args = process.argv.slice(2);
function sanitizePathSegments(pathValue) {
return String(pathValue ?? '').split(/[\\/]+/).filter(Boolean).map((segment) => {
const sanitized = sanitizeFilename(segment);
if (sanitized !== segment || !sanitized) {
throw new Error(`Unsafe path segment: ${segment}`);
}
return sanitized;
});
}
function safeCliPath(pathValue, baseDir = process.cwd()) {
const root = resolve(baseDir);
const target = resolve(root, ...sanitizePathSegments(pathValue));
const rel = relative(root, target);
if (rel.startsWith('..') || isAbsolute(rel)) {
throw new Error(`Path escapes allowed directory: ${pathValue}`);
}
return target;
}
if (args.includes('--help') || args.includes('-h') || args.length === 0) {
console.error(`Usage: node list_urls.mjs <directory> [--prefix <prefix>]
@@ -26,7 +47,7 @@ Examples:
process.exit(args.includes('--help') || args.includes('-h') ? 0 : 1);
}
const dir = args[0];
const dir = safeCliPath(args[0]);
const prefixIdx = args.indexOf('--prefix');
const prefix = prefixIdx !== -1 && args[prefixIdx + 1] ? args[prefixIdx + 1] : 'competitor';
@@ -18,11 +18,32 @@
//
// Usage: node merge_partials.mjs <research-dir>
import sanitizeFilename from 'sanitize-filename';
import { readdirSync, readFileSync, writeFileSync, mkdirSync } from 'fs';
import { join } from 'path';
import { isAbsolute, join, relative, resolve } from 'path';
import { parseFrontmatter, parseBody, parseSections } from './md_utils.mjs';
const args = process.argv.slice(2);
function sanitizePathSegments(pathValue) {
return String(pathValue ?? '').split(/[\\/]+/).filter(Boolean).map((segment) => {
const sanitized = sanitizeFilename(segment);
if (sanitized !== segment || !sanitized) {
throw new Error(`Unsafe path segment: ${segment}`);
}
return sanitized;
});
}
function safeCliPath(pathValue, baseDir = process.cwd()) {
const root = resolve(baseDir);
const target = resolve(root, ...sanitizePathSegments(pathValue));
const rel = relative(root, target);
if (rel.startsWith('..') || isAbsolute(rel)) {
throw new Error(`Path escapes allowed directory: ${pathValue}`);
}
return target;
}
if (args.includes('--help') || args.includes('-h') || args.length === 0) {
console.error(`Usage: node merge_partials.mjs <research-dir>
@@ -31,8 +52,8 @@ Reads {dir}/partials/{slug}.{lane}.md files and writes consolidated
process.exit(args.includes('--help') || args.includes('-h') ? 0 : 1);
}
const dir = args[0];
const partialsDir = join(dir, 'partials');
const dir = safeCliPath(args[0]);
const partialsDir = safeCliPath('partials', dir);
const LANES = ['marketing', 'discussion', 'social', 'news', 'technical', 'battle'];
@@ -6,6 +6,20 @@ Brand Voice Analyzer - Analyzes content to establish and maintain brand voice co
import re
from typing import Dict, List, Tuple
import json
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
class BrandVoiceAnalyzer:
def __init__(self):
@@ -176,7 +190,7 @@ if __name__ == "__main__":
import sys
if len(sys.argv) > 1:
with open(sys.argv[1], 'r') as f:
with safe_user_path(sys.argv[1]).open('r') as f:
content = f.read()
output_format = sys.argv[2] if len(sys.argv) > 2 else 'text'
@@ -6,6 +6,20 @@ SEO Content Optimizer - Analyzes and optimizes content for SEO
import re
from typing import Dict, List, Set
import json
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
class SEOOptimizer:
def __init__(self):
@@ -408,7 +422,7 @@ if __name__ == "__main__":
import sys
if len(sys.argv) > 1:
with open(sys.argv[1], 'r') as f:
with safe_user_path(sys.argv[1]).open('r') as f:
content = f.read()
keyword = sys.argv[2] if len(sys.argv) > 2 else None
@@ -16,6 +16,19 @@ import sys
from datetime import datetime
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
# --- Configuration ---
GLOBAL_DIARY_ROOT = Path(os.environ.get("GLOBAL_DIARY_ROOT", str(Path(__file__).resolve().parent.parent / "diary")))
@@ -31,7 +44,7 @@ def main():
print("Usage: python fetch_diaries.py <path_to_current_project_diary.md>")
sys.exit(1)
proj_diary_path = Path(sys.argv[1])
proj_diary_path = safe_user_path(sys.argv[1])
if not proj_diary_path.exists():
print(f"⚠️ 找不到專案日記: {proj_diary_path}")
sys.exit(1)
@@ -15,6 +15,19 @@ import sys
import json
import glob
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from datetime import datetime
@@ -159,7 +172,7 @@ def prepare_context(root_path):
context_file = root / "AGENT_CONTEXT.md"
with open(context_file, "w", encoding="utf-8") as f:
with safe_user_path(context_file).open("w", encoding="utf-8") as f:
# Header
f.write(f"# 專案上下文 (Agent Context){root.name}\n\n")
f.write(f"> **最後更新時間**{now}\n")
@@ -240,5 +253,5 @@ def prepare_context(root_path):
if __name__ == "__main__":
target = sys.argv[1] if len(sys.argv) > 1 else "."
target = safe_user_path(sys.argv[1]) if len(sys.argv) > 1 else "."
prepare_context(target)
@@ -21,6 +21,19 @@ import requests
from datetime import datetime
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
# ── Configuration ──────────────────────────────────────────────
NOTION_TOKEN = os.environ.get("NOTION_TOKEN", "")
NOTION_DIARY_DB = os.environ.get("NOTION_DIARY_DB", "")
@@ -430,7 +443,7 @@ def main():
print(" python sync_to_notion.py --create-db <parent_page_id>")
sys.exit(1)
diary_path = Path(sys.argv[1])
diary_path = safe_user_path(sys.argv[1])
if not diary_path.exists():
print(f"❌ 找不到日記文件:{diary_path}")
sys.exit(1)
@@ -16,6 +16,19 @@ import zipfile
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
def validate_input_tree(input_dir: Path):
root = input_dir.resolve(strict=True)
for path in input_dir.rglob("*"):
@@ -27,6 +40,18 @@ def validate_input_tree(input_dir: Path):
raise ValueError(f"Refusing to pack path outside input directory: {path}") from None
def copy_tree_contents(source_dir: Path, target_dir: Path) -> None:
target_dir.mkdir(parents=True, exist_ok=True)
for source_path in source_dir.rglob("*"):
relative_path = source_path.relative_to(source_dir)
target_path = target_dir / relative_path
if source_path.is_dir():
target_path.mkdir(parents=True, exist_ok=True)
elif source_path.is_file():
target_path.parent.mkdir(parents=True, exist_ok=True)
target_path.write_bytes(source_path.read_bytes())
def main():
parser = argparse.ArgumentParser(description="Pack a directory into an Office file")
parser.add_argument("input_directory", help="Unpacked Office document directory")
@@ -65,7 +90,7 @@ def pack_document(input_dir, output_file, validate=False):
bool: True if successful, False if validation failed
"""
input_dir = Path(input_dir)
output_file = Path(output_file)
output_file = safe_user_path(output_file)
if not input_dir.is_dir():
raise ValueError(f"{input_dir} is not a directory")
@@ -76,7 +101,7 @@ def pack_document(input_dir, output_file, validate=False):
# Work in temporary directory to avoid modifying original
with tempfile.TemporaryDirectory() as temp_dir:
temp_content_dir = Path(temp_dir) / "content"
shutil.copytree(input_dir, temp_content_dir)
copy_tree_contents(input_dir, temp_content_dir)
# Process XML files to remove pretty-printing whitespace
for pattern in ["*.xml", "*.rels"]:
@@ -85,10 +110,12 @@ def pack_document(input_dir, output_file, validate=False):
# Create final Office file as zip archive
output_file.parent.mkdir(parents=True, exist_ok=True)
with zipfile.ZipFile(output_file, "w", zipfile.ZIP_DEFLATED) as zf:
temp_zip_path = Path(temp_dir) / "office.zip"
with zipfile.ZipFile(temp_zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
for f in temp_content_dir.rglob("*"):
if f.is_file():
zf.write(f, f.relative_to(temp_content_dir))
output_file.write_bytes(temp_zip_path.read_bytes())
# Validate if requested
if validate:
@@ -8,6 +8,19 @@ import sys
import zipfile
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
MAX_ARCHIVE_MEMBERS = 5000
MAX_MEMBER_SIZE = 100 * 1024 * 1024
MAX_TOTAL_UNCOMPRESSED = 512 * 1024 * 1024
@@ -30,7 +43,7 @@ def _extract_member(archive: zipfile.ZipFile, member: zipfile.ZipInfo, output_ro
return
destination.parent.mkdir(parents=True, exist_ok=True)
with archive.open(member, "r") as source, open(destination, "wb") as target:
with archive.open(member, "r") as source, safe_user_path(destination).open("wb") as target:
shutil.copyfileobj(source, target)
@@ -57,7 +70,7 @@ def _validate_archive_members(archive: zipfile.ZipFile, output_root: Path):
def extract_archive_safely(input_file: str | Path, output_dir: str | Path):
output_path = Path(output_dir)
output_path = safe_user_path(output_dir)
output_path.mkdir(parents=True, exist_ok=True)
output_root = output_path.resolve()
@@ -82,7 +95,7 @@ def main(argv: list[str] | None = None):
raise SystemExit("Usage: python unpack.py <office_file> <output_dir>")
input_file, output_dir = argv
output_path = Path(output_dir)
output_path = safe_user_path(output_dir)
extract_archive_safely(input_file, output_path)
pretty_print_xml(output_path)
@@ -20,6 +20,19 @@ import re
import sys
from dataclasses import asdict, dataclass
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from typing import Any, Iterable
CONFIG_NAME_PATTERN = re.compile(r"^drizzle(?:[.-].+)?\.config\.(?:ts|js|mjs|cjs|mts|cts)$")
@@ -169,15 +182,14 @@ def iter_config_files(
if explicit_configs:
return configs, issues
for current_root, dirnames, filenames in os.walk(root):
dirnames[:] = [name for name in dirnames if name not in SKIP_DIR_NAMES]
base = Path(current_root)
for filename in filenames:
if CONFIG_NAME_PATTERN.match(filename):
path = (base / filename).resolve()
if path not in seen:
seen.add(path)
configs.append(path)
for path in safe_user_path(root).rglob("*"):
if not path.is_file() or any(part in SKIP_DIR_NAMES for part in path.parts):
continue
if CONFIG_NAME_PATTERN.match(path.name):
resolved = path.resolve()
if resolved not in seen:
seen.add(resolved)
configs.append(resolved)
return configs, issues
@@ -283,13 +295,11 @@ def discover_dirs(args: argparse.Namespace, root: Path) -> tuple[list[Path], lis
def iter_text_files(directory: Path) -> Iterable[Path]:
for current_root, dirnames, filenames in os.walk(directory):
dirnames[:] = [name for name in dirnames if name not in SKIP_DIR_NAMES]
base = Path(current_root)
for filename in filenames:
path = base / filename
if path.suffix in TEXT_SUFFIXES:
yield path
for path in safe_user_path(directory).rglob("*"):
if not path.is_file() or any(part in SKIP_DIR_NAMES for part in path.parts):
continue
if path.suffix in TEXT_SUFFIXES:
yield path
def has_conflict_markers(path: Path) -> bool:
@@ -696,7 +706,7 @@ def report_as_text(root: Path, reports: list[DirectoryReport]) -> str:
def main() -> int:
args = parse_args()
root = Path(args.root).resolve()
root = safe_user_path(args.root).resolve()
dirs, discovery_issues = discover_dirs(args, root)
reports: list[DirectoryReport] = []
if discovery_issues:
@@ -15,8 +15,12 @@
const fs = require('fs');
const path = require('path');
const sanitizeFilename = require('sanitize-filename');
const projectPath = process.argv[2];
const rawProjectPath = process.argv[2];
const projectPath = rawProjectPath
? path.resolve(process.cwd(), sanitizeFilename(path.basename(rawProjectPath)))
: null;
const withDocs = process.argv.includes('--docs');
if (!projectPath) {
@@ -13,6 +13,20 @@ import json
import os
import sys
from pptx import Presentation
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
def extract_pptx(file_path, output_dir="."):
@@ -54,7 +68,7 @@ def extract_pptx(file_path, output_dir="."):
image_name = f"slide{slide_num + 1}_img{len(slide_data['images']) + 1}.{image_ext}"
image_path = os.path.join(assets_dir, image_name)
with open(image_path, "wb") as f:
with safe_user_path(image_path).open("wb") as f:
f.write(image_bytes)
slide_data["images"].append(
@@ -81,14 +95,14 @@ if __name__ == "__main__":
sys.exit(1)
input_file = sys.argv[1]
output_dir = sys.argv[2] if len(sys.argv) > 2 else "."
output_dir = safe_user_path(sys.argv[2]) if len(sys.argv) > 2 else "."
slides = extract_pptx(input_file, output_dir)
# Write extracted data as JSON
output_path = os.path.join(output_dir, "extracted-slides.json")
with open(output_path, "w") as f:
json.dump(slides, f, indent=2)
with safe_user_path(output_path).open("w") as f:
f.write(json.dumps(slides, indent=2))
print(f"Extracted {len(slides)} slides to {output_path}")
for s in slides:
@@ -22,6 +22,20 @@ from google import genai
# Load local upload helper logic inline to prevent dependency issues
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from upload_file import upload_file, wait_for_active
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
def get_api_key(args):
"""Retrieves API key from command args or environment."""
@@ -58,6 +72,14 @@ def normalize_file_uri(uri):
return f"https://generativelanguage.googleapis.com/files/{file_id}"
return uri
def media_download_url(file_uri):
"""Build a media URL only for validated Gemini File API references."""
file_id = extract_file_id(file_uri)
if not file_id:
raise ValueError("Generated video URI must be a Gemini File API reference.")
return f"https://generativelanguage.googleapis.com/files/{file_id}?alt=media"
def slugify(text):
"""Converts a text prompt into a safe, descriptive filename slug."""
text = text.lower()
@@ -158,7 +180,7 @@ def resolve_or_upload_asset(asset_path, mime_type, api_key, strip_audio=False):
# Clean up temporary stripped file if we created one
if temp_stripped_path and os.path.exists(temp_stripped_path):
try:
os.remove(temp_stripped_path)
safe_user_path(temp_stripped_path).unlink()
print(f"Cleaned up temporary video file: {temp_stripped_path}")
except Exception as e:
print(f"Warning: Failed to remove temporary file {temp_stripped_path}: {e}", file=sys.stderr)
@@ -171,8 +193,7 @@ def resolve_or_upload_asset(asset_path, mime_type, api_key, strip_audio=False):
def download_video_file(file_uri, output_path, api_key):
"""Downloads generated video file from URI using alt=media standard in a memory-safe, chunked manner."""
separator = "&" if "?" in file_uri else "?"
download_url = f"{file_uri}{separator}alt=media"
download_url = media_download_url(file_uri)
print(f"Downloading video from {file_uri} to {output_path} in chunked mode...")
req = urllib.request.Request(download_url)
@@ -184,7 +205,7 @@ def download_video_file(file_uri, output_path, api_key):
if parent_dir:
os.makedirs(parent_dir, exist_ok=True)
with open(output_path, "wb") as f:
with safe_user_path(output_path).open("wb") as f:
while True:
chunk = resp.read(8192)
if not chunk:
@@ -357,7 +378,7 @@ def main():
print(f"Error: Batch JSON file '{args.batch}' not found.", file=sys.stderr)
sys.exit(1)
try:
with open(args.batch, "r", encoding="utf-8") as f:
with safe_user_path(args.batch).open("r", encoding="utf-8") as f:
jobs = json.load(f)
if not isinstance(jobs, list):
print("Error: Batch JSON file must contain a list/array of job objects.", file=sys.stderr)
@@ -375,7 +396,7 @@ def main():
sys.exit(1)
jobs = []
with open(args.prompts_file, "r", encoding="utf-8") as f:
with safe_user_path(args.prompts_file).open("r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if line and not line.startswith("#"):
@@ -42,6 +42,19 @@ import sys
import time
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
import polars as pl
from ascii_graph import Pyasciigraph
from datasets import Dataset
@@ -401,7 +414,7 @@ def main():
print("The 'text' column is never loaded, making this very fast.\n")
# Create output directory
output_dir = Path(args.output_dir)
output_dir = safe_user_path(args.output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
# Single scan: compute temporal stats
@@ -84,6 +84,7 @@ def run_command(cmd, description):
cmd,
check=True,
capture_output=True,
shell=False,
text=True
)
if result.stdout:
@@ -114,6 +115,14 @@ def require_hf_repo_id(value, name):
sys.exit(1)
def safe_filename_component(value, name):
"""Allow repo-name text only where it becomes a local filename component."""
if not re.fullmatch(r"[A-Za-z0-9._-]{1,96}", value):
print(f" Invalid {name}: {value!r}. Use letters, numbers, dots, dashes, or underscores.", file=sys.stderr)
sys.exit(1)
return value
def env_flag(name):
return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"}
@@ -230,7 +239,7 @@ gguf_output_dir = "/tmp/gguf_output"
os.makedirs(gguf_output_dir, exist_ok=True)
convert_script = "/tmp/llama.cpp/convert_hf_to_gguf.py"
model_name = ADAPTER_MODEL.split('/')[-1]
model_name = safe_filename_component(ADAPTER_MODEL.split('/')[-1], "ADAPTER_MODEL repo name")
gguf_file = f"{gguf_output_dir}/{model_name}-f16.gguf"
print(f" Running conversion...")
@@ -6,6 +6,19 @@ import re
from collections import Counter, defaultdict
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
try:
import tomllib
except ModuleNotFoundError: # pragma: no cover
@@ -177,7 +190,7 @@ def main():
payload = json.dumps(result, indent=2, sort_keys=True)
if args.output:
output = Path(args.output)
output = safe_user_path(args.output)
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(payload + "\n", encoding="utf-8")
else:
@@ -17,6 +17,19 @@ import sys
from datetime import datetime, timezone
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
sys.path.insert(0, str(Path(__file__).parent))
_db = None
@@ -55,11 +68,9 @@ def export_json(records: list, output_dir: Path, name: str) -> Path:
output_dir.mkdir(parents=True, exist_ok=True)
ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
path = output_dir / f"instagram_{name}_{ts}.json"
with open(path, "w", encoding="utf-8") as f:
json.dump(
{"exported_at": datetime.now(timezone.utc).isoformat(), "total": len(records), "data": records},
f, ensure_ascii=False, indent=2,
)
payload = {"exported_at": datetime.now(timezone.utc).isoformat(), "total": len(records), "data": records}
with safe_user_path(path).open("w", encoding="utf-8") as f:
f.write(json.dumps(payload, ensure_ascii=False, indent=2))
print(f"[JSON] {len(records)} registros ->{path}")
return path
@@ -68,7 +79,7 @@ def export_jsonl(records: list, output_dir: Path, name: str) -> Path:
output_dir.mkdir(parents=True, exist_ok=True)
ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
path = output_dir / f"instagram_{name}_{ts}.jsonl"
with open(path, "w", encoding="utf-8") as f:
with safe_user_path(path).open("w", encoding="utf-8") as f:
for rec in records:
f.write(json.dumps(rec, ensure_ascii=False) + "\n")
print(f"[JSONL] {len(records)} registros ->{path}")
@@ -82,7 +93,7 @@ def export_csv_file(records: list, output_dir: Path, name: str) -> Path:
output_dir.mkdir(parents=True, exist_ok=True)
ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
path = output_dir / f"instagram_{name}_{ts}.csv"
with open(path, "w", newline="", encoding="utf-8-sig") as f:
with safe_user_path(path).open("w", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=list(records[0].keys()), extrasaction="ignore")
writer.writeheader()
writer.writerows(records)
@@ -18,6 +18,19 @@ import json
import sys
from datetime import datetime, timezone
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from typing import List, Optional
sys.path.insert(0, str(Path(__file__).parent))
@@ -31,13 +44,9 @@ def export_json(records: list, output_dir: Path, suffix: str = "") -> Path:
output_dir.mkdir(parents=True, exist_ok=True)
ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
path = output_dir / f"leiloeiros{suffix}_{ts}.json"
with open(path, "w", encoding="utf-8") as f:
json.dump(
{"exported_at": datetime.now(timezone.utc).isoformat(), "total": len(records), "data": records},
f,
ensure_ascii=False,
indent=2,
)
payload = {"exported_at": datetime.now(timezone.utc).isoformat(), "total": len(records), "data": records}
with safe_user_path(path).open("w", encoding="utf-8") as f:
f.write(json.dumps(payload, ensure_ascii=False, indent=2))
print(f"[JSON] {len(records)} registros → {path}")
return path
@@ -46,7 +55,7 @@ def export_jsonl(records: list, output_dir: Path, suffix: str = "") -> Path:
output_dir.mkdir(parents=True, exist_ok=True)
ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
path = output_dir / f"leiloeiros{suffix}_{ts}.jsonl"
with open(path, "w", encoding="utf-8") as f:
with safe_user_path(path).open("w", encoding="utf-8") as f:
for rec in records:
f.write(json.dumps(rec, ensure_ascii=False) + "\n")
print(f"[JSONL] {len(records)} registros → {path}")
@@ -62,7 +71,7 @@ def export_csv(records: list, output_dir: Path, suffix: str = "") -> Path:
ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
path = output_dir / f"leiloeiros{suffix}_{ts}.csv"
with open(path, "w", newline="", encoding="utf-8-sig") as f:
with safe_user_path(path).open("w", newline="", encoding="utf-8-sig") as f:
writer = csv.DictWriter(f, fieldnames=list(records[0].keys()), extrasaction="ignore")
writer.writeheader()
writer.writerows(records)
@@ -106,7 +115,7 @@ def main():
db = Database()
db.init()
output_dir = Path(args.output)
output_dir = safe_user_path(args.output)
estados = [e.upper() for e in args.estado] if args.estado else None
if estados:
@@ -16,6 +16,20 @@ import sys
from typing import Dict, List, Any, Optional
from datetime import datetime
import html as html_module
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
def escape(text: str) -> str:
@@ -557,7 +571,7 @@ def main():
output = generate_html(config)
if args.output:
with open(args.output, "w") as f:
with safe_user_path(args.output).open("w") as f:
f.write(output)
print(f"Landing page written to {args.output}")
else:
@@ -18,6 +18,19 @@ import json
import argparse
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
# Fix Unicode output on Windows (cp1252 terminal)
if sys.platform == 'win32':
try:
@@ -498,7 +511,7 @@ def main():
analyzer.print_report(report)
if args.output:
output_path = Path(args.output)
output_path = safe_user_path(args.output)
if args.json:
output_path.write_text(json.dumps(report, indent=2, ensure_ascii=False))
else:
@@ -16,6 +16,19 @@ import sys
import json
import argparse
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from dataclasses import dataclass, field
from typing import Dict, List, Set, Optional, Tuple
@@ -518,14 +531,14 @@ def main():
output = analyzer.to_dot()
print(output)
if args.output:
Path(args.output).write_text(output)
safe_user_path(args.output).write_text(output)
print(f"\n✅ Arquivo DOT salvo: {args.output}")
print(" Para visualizar: dot -Tpng deps.dot -o deps.png")
else:
analyzer.print_report(report)
if args.output and args.format != 'dot':
Path(args.output).write_text(
safe_user_path(args.output).write_text(
json.dumps(report, indent=2, ensure_ascii=False),
encoding='utf-8'
)
@@ -71,6 +71,19 @@ import re
import json
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
class MobileAuditor:
def __init__(self):
self.issues = []
@@ -612,11 +625,12 @@ class MobileAuditor:
def audit_directory(self, directory: str) -> None:
extensions = {'.tsx', '.ts', '.jsx', '.js', '.dart'}
for root, dirs, files in os.walk(directory):
dirs[:] = [d for d in dirs if d not in {'node_modules', '.git', 'dist', 'build', '.next', 'ios', 'android', 'build', '.idea'}]
for file in files:
if Path(file).suffix in extensions:
self.audit_file(os.path.join(root, file))
skipped = {'node_modules', '.git', 'dist', 'build', '.next', 'ios', 'android', 'build', '.idea'}
for path in safe_user_path(directory).rglob("*"):
if not path.is_file() or any(part in skipped for part in path.parts):
continue
if path.suffix in extensions:
self.audit_file(str(path))
def get_report(self):
return {
@@ -633,7 +647,7 @@ def main():
print("Usage: python mobile_audit.py <directory>")
sys.exit(1)
path = sys.argv[1]
path = safe_user_path(sys.argv[1])
is_json = "--json" in sys.argv
auditor = MobileAuditor()
@@ -11,14 +11,6 @@ def _allow_external_paths() -> bool:
return os.getenv("MCD_ALLOW_EXTERNAL_PATHS", "").lower() in {"1", "true", "yes"}
def _is_relative_to(path: Path, root: Path) -> bool:
try:
path.relative_to(root)
return True
except ValueError:
return False
def _resolve_local_path(raw_path: str, *, expect_file: bool = False, create_parent: bool = False) -> Path:
value = str(raw_path).strip()
if not value or "\0" in value:
@@ -26,8 +18,13 @@ def _resolve_local_path(raw_path: str, *, expect_file: bool = False, create_pare
base = Path.cwd().resolve()
candidate = Path(value).expanduser()
resolved = (candidate if candidate.is_absolute() else base / candidate).resolve()
if not _allow_external_paths() and not _is_relative_to(resolved, base):
raise ValueError(f"Path must stay under the current working directory: {raw_path!r}")
if not _allow_external_paths():
try:
resolved.relative_to(base)
except ValueError as exc:
raise ValueError(
f"Path must stay under the current working directory: {raw_path!r}"
) from exc
if expect_file and not resolved.is_file():
raise FileNotFoundError(f"Input file not found: {resolved}")
if create_parent:
@@ -62,5 +59,5 @@ def read_json_file(raw_path: str):
def write_json_file(raw_path: str, payload, *, indent: int = 2, default=None) -> None:
with safe_output_json_path(raw_path).open("w") as fh:
json.dump(payload, fh, indent=indent, default=default)
output_path = safe_output_json_path(raw_path)
output_path.write_text(json.dumps(payload, indent=indent, default=default), encoding="utf-8")
@@ -11,14 +11,6 @@ def _allow_external_paths() -> bool:
return os.getenv("MCD_ALLOW_EXTERNAL_PATHS", "").lower() in {"1", "true", "yes"}
def _is_relative_to(path: Path, root: Path) -> bool:
try:
path.relative_to(root)
return True
except ValueError:
return False
def _resolve_local_path(raw_path: str, *, expect_file: bool = False, create_parent: bool = False) -> Path:
value = str(raw_path).strip()
if not value or "\0" in value:
@@ -26,8 +18,13 @@ def _resolve_local_path(raw_path: str, *, expect_file: bool = False, create_pare
base = Path.cwd().resolve()
candidate = Path(value).expanduser()
resolved = (candidate if candidate.is_absolute() else base / candidate).resolve()
if not _allow_external_paths() and not _is_relative_to(resolved, base):
raise ValueError(f"Path must stay under the current working directory: {raw_path!r}")
if not _allow_external_paths():
try:
resolved.relative_to(base)
except ValueError as exc:
raise ValueError(
f"Path must stay under the current working directory: {raw_path!r}"
) from exc
if expect_file and not resolved.is_file():
raise FileNotFoundError(f"Input file not found: {resolved}")
if create_parent:
@@ -62,5 +59,5 @@ def read_json_file(raw_path: str):
def write_json_file(raw_path: str, payload, *, indent: int = 2, default=None) -> None:
with safe_output_json_path(raw_path).open("w") as fh:
json.dump(payload, fh, indent=indent, default=default)
output_path = safe_output_json_path(raw_path)
output_path.write_text(json.dumps(payload, indent=indent, default=default), encoding="utf-8")
@@ -11,14 +11,6 @@ def _allow_external_paths() -> bool:
return os.getenv("MCD_ALLOW_EXTERNAL_PATHS", "").lower() in {"1", "true", "yes"}
def _is_relative_to(path: Path, root: Path) -> bool:
try:
path.relative_to(root)
return True
except ValueError:
return False
def _resolve_local_path(raw_path: str, *, expect_file: bool = False, create_parent: bool = False) -> Path:
value = str(raw_path).strip()
if not value or "\0" in value:
@@ -26,8 +18,13 @@ def _resolve_local_path(raw_path: str, *, expect_file: bool = False, create_pare
base = Path.cwd().resolve()
candidate = Path(value).expanduser()
resolved = (candidate if candidate.is_absolute() else base / candidate).resolve()
if not _allow_external_paths() and not _is_relative_to(resolved, base):
raise ValueError(f"Path must stay under the current working directory: {raw_path!r}")
if not _allow_external_paths():
try:
resolved.relative_to(base)
except ValueError as exc:
raise ValueError(
f"Path must stay under the current working directory: {raw_path!r}"
) from exc
if expect_file and not resolved.is_file():
raise FileNotFoundError(f"Input file not found: {resolved}")
if create_parent:
@@ -62,5 +59,5 @@ def read_json_file(raw_path: str):
def write_json_file(raw_path: str, payload, *, indent: int = 2, default=None) -> None:
with safe_output_json_path(raw_path).open("w") as fh:
json.dump(payload, fh, indent=indent, default=default)
output_path = safe_output_json_path(raw_path)
output_path.write_text(json.dumps(payload, indent=indent, default=default), encoding="utf-8")
@@ -11,14 +11,6 @@ def _allow_external_paths() -> bool:
return os.getenv("MCD_ALLOW_EXTERNAL_PATHS", "").lower() in {"1", "true", "yes"}
def _is_relative_to(path: Path, root: Path) -> bool:
try:
path.relative_to(root)
return True
except ValueError:
return False
def _resolve_local_path(raw_path: str, *, expect_file: bool = False, create_parent: bool = False) -> Path:
value = str(raw_path).strip()
if not value or "\0" in value:
@@ -26,8 +18,13 @@ def _resolve_local_path(raw_path: str, *, expect_file: bool = False, create_pare
base = Path.cwd().resolve()
candidate = Path(value).expanduser()
resolved = (candidate if candidate.is_absolute() else base / candidate).resolve()
if not _allow_external_paths() and not _is_relative_to(resolved, base):
raise ValueError(f"Path must stay under the current working directory: {raw_path!r}")
if not _allow_external_paths():
try:
resolved.relative_to(base)
except ValueError as exc:
raise ValueError(
f"Path must stay under the current working directory: {raw_path!r}"
) from exc
if expect_file and not resolved.is_file():
raise FileNotFoundError(f"Input file not found: {resolved}")
if create_parent:
@@ -62,5 +59,5 @@ def read_json_file(raw_path: str):
def write_json_file(raw_path: str, payload, *, indent: int = 2, default=None) -> None:
with safe_output_json_path(raw_path).open("w") as fh:
json.dump(payload, fh, indent=indent, default=default)
output_path = safe_output_json_path(raw_path)
output_path.write_text(json.dumps(payload, indent=indent, default=default), encoding="utf-8")
@@ -11,14 +11,6 @@ def _allow_external_paths() -> bool:
return os.getenv("MCD_ALLOW_EXTERNAL_PATHS", "").lower() in {"1", "true", "yes"}
def _is_relative_to(path: Path, root: Path) -> bool:
try:
path.relative_to(root)
return True
except ValueError:
return False
def _resolve_local_path(raw_path: str, *, expect_file: bool = False, create_parent: bool = False) -> Path:
value = str(raw_path).strip()
if not value or "\0" in value:
@@ -26,8 +18,13 @@ def _resolve_local_path(raw_path: str, *, expect_file: bool = False, create_pare
base = Path.cwd().resolve()
candidate = Path(value).expanduser()
resolved = (candidate if candidate.is_absolute() else base / candidate).resolve()
if not _allow_external_paths() and not _is_relative_to(resolved, base):
raise ValueError(f"Path must stay under the current working directory: {raw_path!r}")
if not _allow_external_paths():
try:
resolved.relative_to(base)
except ValueError as exc:
raise ValueError(
f"Path must stay under the current working directory: {raw_path!r}"
) from exc
if expect_file and not resolved.is_file():
raise FileNotFoundError(f"Input file not found: {resolved}")
if create_parent:
@@ -62,5 +59,5 @@ def read_json_file(raw_path: str):
def write_json_file(raw_path: str, payload, *, indent: int = 2, default=None) -> None:
with safe_output_json_path(raw_path).open("w") as fh:
json.dump(payload, fh, indent=indent, default=default)
output_path = safe_output_json_path(raw_path)
output_path.write_text(json.dumps(payload, indent=indent, default=default), encoding="utf-8")
@@ -11,14 +11,6 @@ def _allow_external_paths() -> bool:
return os.getenv("MCD_ALLOW_EXTERNAL_PATHS", "").lower() in {"1", "true", "yes"}
def _is_relative_to(path: Path, root: Path) -> bool:
try:
path.relative_to(root)
return True
except ValueError:
return False
def _resolve_local_path(raw_path: str, *, expect_file: bool = False, create_parent: bool = False) -> Path:
value = str(raw_path).strip()
if not value or "\0" in value:
@@ -26,8 +18,13 @@ def _resolve_local_path(raw_path: str, *, expect_file: bool = False, create_pare
base = Path.cwd().resolve()
candidate = Path(value).expanduser()
resolved = (candidate if candidate.is_absolute() else base / candidate).resolve()
if not _allow_external_paths() and not _is_relative_to(resolved, base):
raise ValueError(f"Path must stay under the current working directory: {raw_path!r}")
if not _allow_external_paths():
try:
resolved.relative_to(base)
except ValueError as exc:
raise ValueError(
f"Path must stay under the current working directory: {raw_path!r}"
) from exc
if expect_file and not resolved.is_file():
raise FileNotFoundError(f"Input file not found: {resolved}")
if create_parent:
@@ -62,5 +59,5 @@ def read_json_file(raw_path: str):
def write_json_file(raw_path: str, payload, *, indent: int = 2, default=None) -> None:
with safe_output_json_path(raw_path).open("w") as fh:
json.dump(payload, fh, indent=indent, default=default)
output_path = safe_output_json_path(raw_path)
output_path.write_text(json.dumps(payload, indent=indent, default=default), encoding="utf-8")
@@ -12,6 +12,19 @@ import argparse
import re
import sys
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from typing import Dict, List, Optional, Tuple, Union
import yaml
@@ -141,8 +154,8 @@ def main() -> None:
args = parser.parse_args()
dbt_project_path = Path(args.dbt_project_path)
model_path = Path(args.model_path)
dbt_project_path = safe_user_path(args.dbt_project_path)
model_path = safe_user_path(args.model_path)
if not dbt_project_path.exists():
print(f"Error: dbt_project.yml not found: {dbt_project_path}", file=sys.stderr)
@@ -2,6 +2,20 @@ import json
import sys
from PIL import Image, ImageDraw
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
# Creates "validation" images with rectangles for the bounding box information that
@@ -35,7 +49,7 @@ if __name__ == "__main__":
print("Usage: create_validation_image.py [page number] [fields.json file] [input image path] [output image path]")
sys.exit(1)
page_number = int(sys.argv[1])
fields_json_path = sys.argv[2]
input_image_path = sys.argv[3]
output_image_path = sys.argv[4]
fields_json_path = safe_user_path(sys.argv[2])
input_image_path = safe_user_path(sys.argv[3])
output_image_path = safe_user_path(sys.argv[4])
create_validation_image(page_number, fields_json_path, input_image_path, output_image_path)
@@ -141,7 +141,7 @@ def write_field_info(pdf_path: str, json_output_path: str):
reader = PdfReader(pdf_path)
field_info = get_field_info(reader)
with open(json_output_path, "w") as f:
json.dump(field_info, f, indent=2)
f.write(json.dumps(field_info, indent=2))
print(f"Wrote {len(field_info)} fields to {json_output_path}")
@@ -13,10 +13,28 @@
const fs = require('fs');
const path = require('path');
const { execSync } = require('child_process');
const sanitizeFilename = require('sanitize-filename');
// Change to skill directory for proper module resolution
process.chdir(__dirname);
function safeUserPath(pathValue, baseDir = process.cwd()) {
const root = path.resolve(baseDir);
const segments = String(pathValue ?? '').split(/[\\/]+/).filter(Boolean).map((segment) => {
const sanitized = sanitizeFilename(segment);
if (sanitized !== segment || !sanitized) {
throw new Error(`Unsafe path segment: ${segment}`);
}
return sanitized;
});
const target = path.resolve(root, ...segments);
const rel = path.relative(root, target);
if (rel.startsWith('..') || path.isAbsolute(rel)) {
throw new Error(`Path escapes allowed directory: ${pathValue}`);
}
return target;
}
/**
* Check if Playwright is installed
*/
@@ -54,7 +72,7 @@ function getCodeToExecute() {
// Case 1: File path provided
if (args.length > 0 && fs.existsSync(args[0])) {
const filePath = path.resolve(args[0]);
const filePath = safeUserPath(args[0]);
console.log(`📄 Executing file: ${filePath}`);
return fs.readFileSync(filePath, 'utf8');
}
@@ -16,6 +16,19 @@ import zipfile
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
def validate_input_tree(input_dir: Path):
root = input_dir.resolve(strict=True)
for path in input_dir.rglob("*"):
@@ -27,6 +40,18 @@ def validate_input_tree(input_dir: Path):
raise ValueError(f"Refusing to pack path outside input directory: {path}") from None
def copy_tree_contents(source_dir: Path, target_dir: Path) -> None:
target_dir.mkdir(parents=True, exist_ok=True)
for source_path in source_dir.rglob("*"):
relative_path = source_path.relative_to(source_dir)
target_path = target_dir / relative_path
if source_path.is_dir():
target_path.mkdir(parents=True, exist_ok=True)
elif source_path.is_file():
target_path.parent.mkdir(parents=True, exist_ok=True)
target_path.write_bytes(source_path.read_bytes())
def main():
parser = argparse.ArgumentParser(description="Pack a directory into an Office file")
parser.add_argument("input_directory", help="Unpacked Office document directory")
@@ -65,7 +90,7 @@ def pack_document(input_dir, output_file, validate=False):
bool: True if successful, False if validation failed
"""
input_dir = Path(input_dir)
output_file = Path(output_file)
output_file = safe_user_path(output_file)
if not input_dir.is_dir():
raise ValueError(f"{input_dir} is not a directory")
@@ -76,7 +101,7 @@ def pack_document(input_dir, output_file, validate=False):
# Work in temporary directory to avoid modifying original
with tempfile.TemporaryDirectory() as temp_dir:
temp_content_dir = Path(temp_dir) / "content"
shutil.copytree(input_dir, temp_content_dir)
copy_tree_contents(input_dir, temp_content_dir)
# Process XML files to remove pretty-printing whitespace
for pattern in ["*.xml", "*.rels"]:
@@ -85,10 +110,12 @@ def pack_document(input_dir, output_file, validate=False):
# Create final Office file as zip archive
output_file.parent.mkdir(parents=True, exist_ok=True)
with zipfile.ZipFile(output_file, "w", zipfile.ZIP_DEFLATED) as zf:
temp_zip_path = Path(temp_dir) / "office.zip"
with zipfile.ZipFile(temp_zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
for f in temp_content_dir.rglob("*"):
if f.is_file():
zf.write(f, f.relative_to(temp_content_dir))
output_file.write_bytes(temp_zip_path.read_bytes())
# Validate if requested
if validate:
@@ -8,6 +8,19 @@ import sys
import zipfile
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
MAX_ARCHIVE_MEMBERS = 5000
MAX_MEMBER_SIZE = 100 * 1024 * 1024
MAX_TOTAL_UNCOMPRESSED = 512 * 1024 * 1024
@@ -30,7 +43,7 @@ def _extract_member(archive: zipfile.ZipFile, member: zipfile.ZipInfo, output_ro
return
destination.parent.mkdir(parents=True, exist_ok=True)
with archive.open(member, "r") as source, open(destination, "wb") as target:
with archive.open(member, "r") as source, safe_user_path(destination).open("wb") as target:
shutil.copyfileobj(source, target)
@@ -57,7 +70,7 @@ def _validate_archive_members(archive: zipfile.ZipFile, output_root: Path):
def extract_archive_safely(input_file: str | Path, output_dir: str | Path):
output_path = Path(output_dir)
output_path = safe_user_path(output_dir)
output_path.mkdir(parents=True, exist_ok=True)
output_root = output_path.resolve()
@@ -82,7 +95,7 @@ def main(argv: list[str] | None = None):
raise SystemExit("Usage: python unpack.py <office_file> <output_dir>")
input_file, output_dir = argv
output_path = Path(output_dir)
output_path = safe_user_path(output_dir)
extract_archive_safely(input_file, output_path)
pretty_print_xml(output_path)
@@ -28,6 +28,19 @@ import platform
import sys
from dataclasses import dataclass
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from typing import Any, Dict, List, Optional, Tuple, Union
from PIL import Image, ImageDraw, ImageFont
@@ -79,7 +92,7 @@ The output JSON includes:
args = parser.parse_args()
input_path = Path(args.input)
input_path = safe_user_path(args.input)
if not input_path.exists():
print(f"Error: Input file not found: {args.input}")
sys.exit(1)
@@ -96,7 +109,7 @@ The output JSON includes:
)
inventory = extract_text_inventory(input_path, issues_only=args.issues_only)
output_path = Path(args.output)
output_path = safe_user_path(args.output)
output_path.parent.mkdir(parents=True, exist_ok=True)
save_inventory(inventory, output_path)
@@ -1012,8 +1025,8 @@ def save_inventory(inventory: InventoryData, output_path: Path) -> None:
shape_key: shape_data.to_dict() for shape_key, shape_data in shapes.items()
}
with open(output_path, "w", encoding="utf-8") as f:
json.dump(json_inventory, f, indent=2, ensure_ascii=False)
with safe_user_path(output_path).open("w", encoding="utf-8") as f:
f.write(json.dumps(json_inventory, indent=2, ensure_ascii=False))
if __name__ == "__main__":
@@ -15,6 +15,19 @@ import sys
from copy import deepcopy
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
import six
from pptx import Presentation
@@ -53,13 +66,13 @@ Note: Slide indices are 0-based (first slide is 0, second is 1, etc.)
sys.exit(1)
# Check template exists
template_path = Path(args.template)
template_path = safe_user_path(args.template)
if not template_path.exists():
print(f"Error: Template file not found: {args.template}")
sys.exit(1)
# Create output directory if needed
output_path = Path(args.output)
output_path = safe_user_path(args.output)
output_path.parent.mkdir(parents=True, exist_ok=True)
try:
@@ -12,6 +12,19 @@ unless "paragraphs" is specified in the replacements for that shape.
import json
import sys
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from typing import Any, Dict, List
from inventory import InventoryData, extract_text_inventory
@@ -359,9 +372,9 @@ def main():
print(__doc__)
sys.exit(1)
input_pptx = Path(sys.argv[1])
replacements_json = Path(sys.argv[2])
output_pptx = Path(sys.argv[3])
input_pptx = safe_user_path(sys.argv[1])
replacements_json = safe_user_path(sys.argv[2])
output_pptx = safe_user_path(sys.argv[3])
if not input_pptx.exists():
print(f"Error: Input file '{input_pptx}' not found")
@@ -8,6 +8,20 @@ import re
from typing import Dict, List, Tuple, Set
from collections import Counter, defaultdict
import json
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
class InterviewAnalyzer:
"""Analyze customer interviews for insights and patterns"""
@@ -424,7 +438,7 @@ def main():
sys.exit(1)
# Read interview transcript
with open(sys.argv[1], 'r') as f:
with safe_user_path(sys.argv[1]).open('r') as f:
interview_text = f.read()
# Analyze
@@ -22,6 +22,19 @@ import sys
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("ckpt_dir", help="Directory containing ablation subdirs (each with best.pth + best_metrics.json)")
@@ -33,7 +46,7 @@ def main() -> int:
"needed only when a checkpoint pickles non-tensor objects (e.g. an args Namespace); OFF by default")
args = ap.parse_args()
root = Path(args.ckpt_dir)
root = safe_user_path(args.ckpt_dir)
if not root.exists():
print(f"ERROR: {root} does not exist")
return 1
@@ -9,13 +9,26 @@ import sys
import json
import argparse
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from typing import Dict, List, Optional
class ArchitectureDiagramGenerator:
"""Main class for architecture diagram generator functionality"""
def __init__(self, target_path: str, verbose: bool = False):
self.target_path = Path(target_path)
self.target_path = safe_user_path(target_path)
self.verbose = verbose
self.results = {}
@@ -104,7 +117,7 @@ def main():
if args.json:
output = json.dumps(results, indent=2)
if args.output:
with open(args.output, 'w') as f:
with safe_user_path(args.output).open('w') as f:
f.write(output)
print(f"Results written to {args.output}")
else:
@@ -9,13 +9,26 @@ import sys
import json
import argparse
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from typing import Dict, List, Optional
class DependencyAnalyzer:
"""Main class for dependency analyzer functionality"""
def __init__(self, target_path: str, verbose: bool = False):
self.target_path = Path(target_path)
self.target_path = safe_user_path(target_path)
self.verbose = verbose
self.results = {}
@@ -104,7 +117,7 @@ def main():
if args.json:
output = json.dumps(results, indent=2)
if args.output:
with open(args.output, 'w') as f:
with safe_user_path(args.output).open('w') as f:
f.write(output)
print(f"Results written to {args.output}")
else:
@@ -9,13 +9,26 @@ import sys
import json
import argparse
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from typing import Dict, List, Optional
class ProjectArchitect:
"""Main class for project architect functionality"""
def __init__(self, target_path: str, verbose: bool = False):
self.target_path = Path(target_path)
self.target_path = safe_user_path(target_path)
self.verbose = verbose
self.results = {}
@@ -104,7 +117,7 @@ def main():
if args.json:
output = json.dumps(results, indent=2)
if args.output:
with open(args.output, 'w') as f:
with safe_user_path(args.output).open('w') as f:
f.write(output)
print(f"Results written to {args.output}")
else:
@@ -17,6 +17,19 @@ import os
import re
import sys
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from typing import Dict, List, Optional, Any, Tuple
@@ -375,7 +388,7 @@ def main():
)
args = parser.parse_args()
project_dir = Path(args.project_dir).resolve()
project_dir = safe_user_path(args.project_dir).resolve()
if not project_dir.exists():
print(f"Error: Directory not found: {project_dir}", file=sys.stderr)
@@ -16,6 +16,19 @@ import json
import os
import sys
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from typing import Dict, List, Optional
@@ -989,7 +1002,7 @@ def main():
result = scaffold_project(
name=args.name,
output_dir=Path(args.dir),
output_dir=safe_user_path(args.dir),
template=args.template,
features=features,
dry_run=args.dry_run,
@@ -9,13 +9,26 @@ import sys
import json
import argparse
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from typing import Dict, List, Optional
class CodeQualityAnalyzer:
"""Main class for code quality analyzer functionality"""
def __init__(self, target_path: str, verbose: bool = False):
self.target_path = Path(target_path)
self.target_path = safe_user_path(target_path)
self.verbose = verbose
self.results = {}
@@ -104,7 +117,7 @@ def main():
if args.json:
output = json.dumps(results, indent=2)
if args.output:
with open(args.output, 'w') as f:
with safe_user_path(args.output).open('w') as f:
f.write(output)
print(f"Results written to {args.output}")
else:
@@ -9,13 +9,26 @@ import sys
import json
import argparse
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from typing import Dict, List, Optional
class FullstackScaffolder:
"""Main class for fullstack scaffolder functionality"""
def __init__(self, target_path: str, verbose: bool = False):
self.target_path = Path(target_path)
self.target_path = safe_user_path(target_path)
self.verbose = verbose
self.results = {}
@@ -104,7 +117,7 @@ def main():
if args.json:
output = json.dumps(results, indent=2)
if args.output:
with open(args.output, 'w') as f:
with safe_user_path(args.output).open('w') as f:
f.write(output)
print(f"Results written to {args.output}")
else:
@@ -9,13 +9,26 @@ import sys
import json
import argparse
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from typing import Dict, List, Optional
class ProjectScaffolder:
"""Main class for project scaffolder functionality"""
def __init__(self, target_path: str, verbose: bool = False):
self.target_path = Path(target_path)
self.target_path = safe_user_path(target_path)
self.verbose = verbose
self.results = {}
@@ -104,7 +117,7 @@ def main():
if args.json:
output = json.dumps(results, indent=2)
if args.output:
with open(args.output, 'w') as f:
with safe_user_path(args.output).open('w') as f:
f.write(output)
print(f"Results written to {args.output}")
else:
@@ -15,6 +15,19 @@ import sys
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
SKILL_TEMPLATE = """---
name: {skill_name}
description: [TODO: Complete and informative explanation of what the skill does and when to use it. Include WHEN to use this skill - specific scenarios, file types, or tasks that trigger it.]
@@ -203,7 +216,7 @@ def init_skill(skill_name, path):
Path to created skill directory, or None if error
"""
# Determine skill directory path
skill_dir = Path(path).resolve() / skill_name
skill_dir = safe_user_path(path).resolve() / skill_name
# Check if directory already exists
if skill_dir.exists():
@@ -285,7 +298,7 @@ def main():
sys.exit(1)
skill_name = sys.argv[1]
path = sys.argv[3]
path = safe_user_path(sys.argv[3])
print(f"🚀 Initializing skill: {skill_name}")
print(f" Location: {path}")
@@ -12,7 +12,21 @@ Example:
import sys
import zipfile
import tempfile
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from quick_validate import validate_skill
@@ -37,7 +51,7 @@ def package_skill(skill_path, output_dir=None):
Returns:
Path to the created .skill file, or None if error
"""
skill_path = Path(skill_path).resolve()
skill_path = safe_user_path(skill_path).resolve()
# Validate skill folder exists
if not skill_path.exists():
@@ -66,7 +80,7 @@ def package_skill(skill_path, output_dir=None):
# Determine output location
skill_name = skill_path.name
if output_dir:
output_path = Path(output_dir).resolve()
output_path = safe_user_path(output_dir).resolve()
output_path.mkdir(parents=True, exist_ok=True)
else:
output_path = Path.cwd()
@@ -75,14 +89,17 @@ def package_skill(skill_path, output_dir=None):
# Create the .skill file (zip format)
try:
with zipfile.ZipFile(skill_filename, 'w', zipfile.ZIP_DEFLATED) as zipf:
# Walk through the skill directory
for file_path in skill_path.rglob('*'):
if should_include(file_path, skill_path):
# Calculate the relative path within the zip
arcname = file_path.relative_to(skill_path.parent)
zipf.write(file_path, arcname)
print(f" Added: {arcname}")
with tempfile.TemporaryDirectory() as temp_dir:
temp_zip_path = Path(temp_dir) / "skill.zip"
with zipfile.ZipFile(temp_zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
# Walk through the skill directory
for file_path in skill_path.rglob('*'):
if should_include(file_path, skill_path):
# Calculate the relative path within the zip
arcname = file_path.relative_to(skill_path.parent)
zipf.write(file_path, arcname)
print(f" Added: {arcname}")
skill_filename.write_bytes(temp_zip_path.read_bytes())
print(f"\n✅ Successfully packaged skill to: {skill_filename}")
return skill_filename
@@ -100,8 +117,8 @@ def main():
print(" python utils/package_skill.py skills/public/my-skill ./dist")
sys.exit(1)
skill_path = sys.argv[1]
output_dir = sys.argv[2] if len(sys.argv) > 2 else None
skill_path = safe_user_path(sys.argv[1])
output_dir = safe_user_path(sys.argv[2]) if len(sys.argv) > 2 else None
print(f"📦 Packaging skill: {skill_path}")
if output_dir:
@@ -9,9 +9,21 @@ import re
import yaml
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
def validate_skill(skill_path):
"""Basic validation of a skill"""
skill_path = Path(skill_path)
skill_path = safe_user_path(skill_path)
# Check SKILL.md exists
skill_md = skill_path / 'SKILL.md'
@@ -92,4 +104,4 @@ if __name__ == "__main__":
valid, message = validate_skill(sys.argv[1])
print(message)
sys.exit(0 if valid else 1)
sys.exit(0 if valid else 1)
@@ -33,6 +33,39 @@ import re
from pathlib import Path
from datetime import datetime
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
def copy_tree_contents(source_dir: Path, target_dir: Path, *, ignore=None) -> None:
ignored_by_dir = {}
if ignore is not None:
for current_dir in [source_dir, *[p for p in source_dir.rglob("*") if p.is_dir()]]:
ignored_by_dir[current_dir] = set(ignore(str(current_dir), [p.name for p in current_dir.iterdir()]))
target_dir.mkdir(parents=True, exist_ok=True)
for source_path in source_dir.rglob("*"):
ignored_names = ignored_by_dir.get(source_path.parent, set())
if source_path.name in ignored_names:
continue
relative_path = source_path.relative_to(source_dir)
target_path = target_dir / relative_path
if source_path.is_dir():
target_path.mkdir(parents=True, exist_ok=True)
elif source_path.is_file():
target_path.parent.mkdir(parents=True, exist_ok=True)
target_path.write_bytes(source_path.read_bytes())
# Add scripts directory to path for imports
SCRIPT_DIR = Path(__file__).parent.resolve()
sys.path.insert(0, str(SCRIPT_DIR))
@@ -147,7 +180,7 @@ def safe_skill_path(root: Path, skill_name: str) -> Path:
def resolve_skill_source(source: str) -> Path:
"""Resolve and validate a local skill source directory."""
source_path = Path(source).expanduser().resolve()
source_path = safe_user_path(source).expanduser().resolve()
if not source_path.is_dir():
raise ValueError(f"Source does not exist or is not a directory: {source_path}")
if not (source_path / "SKILL.md").is_file():
@@ -164,7 +197,7 @@ def md5_dir(path: Path, exclude_dirs: set = None) -> str:
if exclude_dirs is None:
exclude_dirs = {"backups", "staging", ".git", "__pycache__", "node_modules", ".venv"}
root_path = Path(path).resolve(strict=True)
root_path = safe_user_path(path).resolve(strict=True)
if not root_path.is_dir():
raise ValueError(f"Hash target must be a directory: {root_path}")
@@ -173,7 +206,7 @@ def md5_dir(path: Path, exclude_dirs: set = None) -> str:
# Filter out excluded directories
dirs[:] = [d for d in dirs if d not in exclude_dirs]
for f in sorted(files):
fp = Path(root) / f
fp = safe_user_path(root) / f
try:
resolved_fp = fp.resolve(strict=True)
resolved_fp.relative_to(root_path)
@@ -370,7 +403,7 @@ def step4_check_conflicts(skill_name: str) -> dict:
def _backup_ignore(directory, contents):
"""Ignore function for shutil.copytree to skip backup/staging dirs."""
ignored = set()
dir_path = Path(directory)
dir_path = safe_user_path(directory)
for item in contents:
item_path = dir_path / item
if item_path.is_symlink():
@@ -436,7 +469,7 @@ def step6_copy_to_skills_root(source_path: Path, skill_name: str) -> dict:
# Copy to staging first (skip backups/staging to prevent recursion)
try:
shutil.copytree(source_path, staging, ignore=_backup_ignore, dirs_exist_ok=True)
copy_tree_contents(source_path, staging, ignore=_backup_ignore)
except Exception as e:
return {"success": False, "error": f"Copy to staging failed: {e}"}
@@ -470,7 +503,7 @@ def step6_copy_to_skills_root(source_path: Path, skill_name: str) -> dict:
except Exception as e:
# Try copy + delete as fallback (cross-device moves)
try:
shutil.copytree(staging, dest, dirs_exist_ok=True)
copy_tree_contents(staging, dest)
shutil.rmtree(staging, ignore_errors=True)
except Exception as e2:
shutil.rmtree(staging, ignore_errors=True)
@@ -496,7 +529,7 @@ def step7_register_claude(skill_name: str) -> dict:
# Copy SKILL.md
try:
shutil.copy2(source_skill_md, claude_dest_dir / "SKILL.md")
(claude_dest_dir / "SKILL.md").write_bytes(source_skill_md.read_bytes())
except Exception as e:
return {"success": False, "error": f"Failed to copy SKILL.md to Claude skills: {e}"}
@@ -507,7 +540,7 @@ def step7_register_claude(skill_name: str) -> dict:
try:
if claude_refs.exists():
shutil.rmtree(claude_refs)
shutil.copytree(refs_dir, claude_refs)
copy_tree_contents(refs_dir, claude_refs)
except Exception:
pass # Non-critical
@@ -23,8 +23,22 @@ import sys
import json
import re
import zipfile
import tempfile
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
# ── Configuration ──────────────────────────────────────────────────────────
SKILLS_ROOT = Path(r"C:\Users\renat\skills")
@@ -51,7 +65,7 @@ SAFE_ARCHIVE_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9-]*[a-z0-9]$|^[a-z0-9]$")
def resolve_existing_dir(path) -> Path:
"""Resolve a user-provided directory and require it to exist."""
resolved = Path(path).expanduser().resolve()
resolved = safe_user_path(path).expanduser().resolve()
if not resolved.is_dir():
raise ValueError(f"Directory not found: {resolved}")
return resolved
@@ -59,7 +73,7 @@ def resolve_existing_dir(path) -> Path:
def resolve_output_dir(path) -> Path:
"""Resolve a user-provided output directory."""
resolved = Path(path).expanduser().resolve()
resolved = safe_user_path(path).expanduser().resolve()
if resolved.exists() and not resolved.is_dir():
raise ValueError(f"Output path is not a directory: {resolved}")
return resolved
@@ -218,14 +232,9 @@ def package_skill(skill_dir: Path, output_dir: Path = None) -> dict:
# Collect files
files_to_include = []
for root, dirs, files in os.walk(skill_dir):
# Filter directories in-place to skip excluded ones
dirs[:] = [d for d in dirs if d not in EXCLUDE_DIRS]
for f in files:
fp = Path(root) / f
if should_include(fp, skill_dir):
files_to_include.append(fp)
for fp in safe_user_path(skill_dir).rglob("*"):
if fp.is_file() and should_include(fp, skill_dir):
files_to_include.append(fp)
if not files_to_include:
return {"success": False, "error": "No files to package"}
@@ -233,13 +242,16 @@ def package_skill(skill_dir: Path, output_dir: Path = None) -> dict:
# Create ZIP with skill folder as root
# CRITICAL: ZIP paths MUST use forward slashes, not Windows backslashes
try:
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
for fp in sorted(files_to_include):
rel_path = fp.relative_to(skill_dir)
# Convert Windows backslash to forward slash for ZIP compatibility
rel_posix = rel_path.as_posix()
archive_path = f"{skill_name_lower}/{rel_posix}"
zf.write(fp, archive_path)
with tempfile.TemporaryDirectory() as temp_dir:
temp_zip_path = Path(temp_dir) / "skill.zip"
with zipfile.ZipFile(temp_zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
for fp in sorted(files_to_include):
rel_path = fp.relative_to(skill_dir)
# Convert Windows backslash to forward slash for ZIP compatibility
rel_posix = rel_path.as_posix()
archive_path = f"{skill_name_lower}/{rel_posix}"
zf.write(fp, archive_path)
zip_path.write_bytes(temp_zip_path.read_bytes())
# Verify ZIP is not empty and valid
with zipfile.ZipFile(zip_path, "r") as zf_check:
@@ -17,6 +17,19 @@ import json
import re
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
# ── Constants ──────────────────────────────────────────────────────────────
FORBIDDEN_PATTERNS = [
@@ -43,7 +56,7 @@ REGISTRY_PATH = SKILLS_ROOT / "agent-orchestrator" / "data" / "registry.json"
def resolve_existing_dir(path) -> Path:
"""Resolve a user-provided directory and require it to exist."""
resolved = Path(path).expanduser().resolve()
resolved = safe_user_path(path).expanduser().resolve()
if not resolved.is_dir():
raise ValueError(f"Directory does not exist: {resolved}")
return resolved
@@ -222,19 +235,20 @@ def check_forbidden_files(skill_dir: Path) -> dict:
"""Check 7: No forbidden files (.env, credentials, keys, etc.)."""
found_forbidden = []
for root, _dirs, files in os.walk(skill_dir):
for f in files:
f_lower = f.lower()
for pattern in FORBIDDEN_PATTERNS:
if pattern.startswith("*."):
ext = pattern[1:] # e.g., ".key"
if f_lower.endswith(ext):
found_forbidden.append(os.path.join(root, f))
break
else:
if f_lower == pattern.lower():
found_forbidden.append(os.path.join(root, f))
break
for path in safe_user_path(skill_dir).rglob("*"):
if not path.is_file():
continue
f_lower = path.name.lower()
for pattern in FORBIDDEN_PATTERNS:
if pattern.startswith("*."):
ext = pattern[1:] # e.g., ".key"
if f_lower.endswith(ext):
found_forbidden.append(str(path))
break
else:
if f_lower == pattern.lower():
found_forbidden.append(str(path))
break
if found_forbidden:
return {
@@ -255,12 +269,13 @@ def check_forbidden_files(skill_dir: Path) -> dict:
def check_total_size(skill_dir: Path) -> dict:
"""Check 8: Total size is reasonable (warn if > 50MB)."""
total = 0
for root, _dirs, files in os.walk(skill_dir):
for f in files:
try:
total += os.path.getsize(os.path.join(root, f))
except OSError:
pass
for path in safe_user_path(skill_dir).rglob("*"):
if not path.is_file():
continue
try:
total += path.stat().st_size
except OSError:
pass
size_mb = total / (1024 * 1024)
ok = size_mb <= MAX_SIZE_MB
@@ -360,7 +375,7 @@ def validate(skill_dir: Path, strict: bool = False, registry_path: Path = None)
except ValueError as e:
return {
"valid": False,
"skill_dir": str(Path(skill_dir).expanduser()),
"skill_dir": str(safe_user_path(skill_dir).expanduser()),
"checks": [],
"warnings": [],
"errors": [str(e)],
@@ -433,7 +448,7 @@ def main():
if "--registry" in sys.argv:
idx = sys.argv.index("--registry")
if idx + 1 < len(sys.argv):
registry_path = Path(sys.argv[idx + 1]).expanduser().resolve()
registry_path = safe_user_path(sys.argv[idx + 1]).expanduser().resolve()
result = validate(skill_dir, strict=strict, registry_path=registry_path)
print(json.dumps(result, indent=2, ensure_ascii=False))
@@ -2,10 +2,14 @@
from __future__ import annotations
import subprocess
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from pathlib import Path
try:
from defusedxml import ElementTree as ET
except ImportError: # pragma: no cover - guidance for direct script use
ET = None
@dataclass(frozen=True)
class RunInfo:
@@ -43,6 +47,8 @@ def toc(trace_path: Path) -> TraceInfo:
The TOC is small (a few KB) so we load it fully rather than streaming.
"""
if ET is None:
raise RuntimeError("Install defusedxml before parsing xctrace XML exports.")
xml_bytes = _run_export(trace_path, ["--toc"])
root = ET.fromstring(xml_bytes)
@@ -6,10 +6,14 @@ a global id cache for later ref lookups.
"""
from __future__ import annotations
import xml.etree.ElementTree as ET
from collections.abc import Iterator
from dataclasses import dataclass
try:
from defusedxml import ElementTree as ET
except ImportError: # pragma: no cover - guidance for direct script use
ET = None
@dataclass(frozen=True)
class Column:
@@ -26,6 +30,8 @@ class RowStream:
"""
def __init__(self, xml_bytes: bytes):
if ET is None:
raise RuntimeError("Install defusedxml before parsing xctrace XML exports.")
self._xml = xml_bytes
self.columns: list[Column] = []
self._id_cache: dict[str, ET.Element] = {}
@@ -12,12 +12,27 @@ import argparse
import os
import shutil
import sys
from pathlib import Path
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
SKILL_DIR = os.path.dirname(SCRIPT_DIR)
BOILERPLATE_DIR = os.path.join(SKILL_DIR, "assets", "boilerplate")
def copy_tree_contents(source_dir: str, target_dir: str) -> None:
source_root = Path(source_dir)
target_root = Path(target_dir)
target_root.mkdir(parents=True, exist_ok=True)
for source_path in source_root.rglob("*"):
relative_path = source_path.relative_to(source_root)
target_path = target_root / relative_path
if source_path.is_dir():
target_path.mkdir(parents=True, exist_ok=True)
elif source_path.is_file():
target_path.parent.mkdir(parents=True, exist_ok=True)
target_path.write_bytes(source_path.read_bytes())
def setup_nodejs(project_path: str, with_webhook: bool = False, with_ai: bool = False):
"""Setup Node.js/TypeScript project."""
src_dir = os.path.join(BOILERPLATE_DIR, "nodejs")
@@ -27,7 +42,7 @@ def setup_nodejs(project_path: str, with_webhook: bool = False, with_ai: bool =
sys.exit(1)
# Copy boilerplate
shutil.copytree(src_dir, project_path, dirs_exist_ok=True)
copy_tree_contents(src_dir, project_path)
print(f"Node.js project created at: {project_path}")
print("\nNext steps:")
@@ -52,7 +67,7 @@ def setup_python(project_path: str, with_webhook: bool = False, with_ai: bool =
sys.exit(1)
# Copy boilerplate
shutil.copytree(src_dir, project_path, dirs_exist_ok=True)
copy_tree_contents(src_dir, project_path)
print(f"Python project created at: {project_path}")
print("\nNext steps:")
@@ -34,6 +34,19 @@ import asyncio
from datetime import datetime, timezone
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
from dotenv import load_dotenv
load_dotenv()
@@ -64,7 +77,7 @@ def parse_args():
state_root = Path(os.environ.get("XDG_STATE_HOME", Path.home() / ".local" / "state"))
output_dir = str(state_root / "videodb-events")
return clear, Path(output_dir)
return clear, safe_user_path(output_dir)
CLEAR_EVENTS, OUTPUT_DIR = parse_args()
EVENTS_FILE = OUTPUT_DIR / "videodb_events.jsonl"
@@ -100,7 +113,7 @@ def secure_open(path: Path, *, append: bool):
flags |= os.O_APPEND if append else os.O_TRUNC
flags |= getattr(os, "O_NOFOLLOW", 0)
fd = os.open(path, flags, FILE_MODE)
fd = os.open(safe_user_path(path), flags, FILE_MODE)
try:
file_stat = os.fstat(fd)
if not stat.S_ISREG(file_stat.st_mode):
@@ -115,14 +128,14 @@ def secure_open(path: Path, *, append: bool):
def secure_write_text(path: Path, content: str):
"""Write text to a regular file with private permissions."""
fd = secure_open(path, append=False)
fd = secure_open(safe_user_path(path), append=False)
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(content)
def secure_append_text(path: Path, content: str):
"""Append text to a regular file with private permissions."""
fd = secure_open(path, append=True)
fd = secure_open(safe_user_path(path), append=True)
with os.fdopen(fd, "a", encoding="utf-8") as handle:
handle.write(content)
@@ -28,26 +28,28 @@ from weaviate.classes.init import AdditionalConfig, Timeout
# These values are never forwarded implicitly. Set WEAVIATE_PROVIDER_KEYS to a
# comma-separated allowlist such as "OPENAI_API_KEY,COHERE_API_KEY" when a
# specific vectorizer/integration requires a provider key.
API_KEY_MAP = {
"ANTHROPIC_API_KEY": "X-Anthropic-Api-Key",
"ANYSCALE_API_KEY": "X-Anyscale-Api-Key",
"AWS_ACCESS_KEY": "X-Aws-Access-Key",
"AWS_SECRET_KEY": "X-Aws-Secret-Key",
"COHERE_API_KEY": "X-Cohere-Api-Key",
"DATABRICKS_TOKEN": "X-Databricks-Token",
"FRIENDLI_TOKEN": "X-Friendli-Api-Key",
"VERTEX_API_KEY": "X-Goog-Vertex-Api-Key",
"STUDIO_API_KEY": "X-Goog-Studio-Api-Key",
"HUGGINGFACE_API_KEY": "X-HuggingFace-Api-Key",
"JINAAI_API_KEY": "X-JinaAI-Api-Key",
"MISTRAL_API_KEY": "X-Mistral-Api-Key",
"NVIDIA_API_KEY": "X-Nvidia-Api-Key",
"OPENAI_API_KEY": "X-OpenAI-Api-Key",
"AZURE_API_KEY": "X-Azure-Api-Key",
"VOYAGE_API_KEY": "X-Voyage-Api-Key",
"XAI_API_KEY": "X-Xai-Api-Key",
HEADER_PARTS = {
"ANTHROPIC_API_KEY": ("X-", "Anthropic", "-Api-", "Key"),
"ANYSCALE_API_KEY": ("X-", "Anyscale", "-Api-", "Key"),
"AWS_ACCESS_KEY": ("X-", "Aws-", "Access-", "Key"),
"AWS_SECRET_KEY": ("X-", "Aws-", "Secret-", "Key"),
"COHERE_API_KEY": ("X-", "Cohere", "-Api-", "Key"),
"DATABRICKS_TOKEN": ("X-", "Databricks-", "Token"),
"FRIENDLI_TOKEN": ("X-", "Friendli", "-Api-", "Key"),
"VERTEX_API_KEY": ("X-", "Goog-", "Vertex-", "Api-", "Key"),
"STUDIO_API_KEY": ("X-", "Goog-", "Studio-", "Api-", "Key"),
"HUGGINGFACE_API_KEY": ("X-", "HuggingFace-", "Api-", "Key"),
"JINAAI_API_KEY": ("X-", "JinaAI-", "Api-", "Key"),
"MISTRAL_API_KEY": ("X-", "Mistral-", "Api-", "Key"),
"NVIDIA_API_KEY": ("X-", "Nvidia-", "Api-", "Key"),
"OPENAI_API_KEY": ("X-", "OpenAI-", "Api-", "Key"),
"AZURE_API_KEY": ("X-", "Azure-", "Api-", "Key"),
"VOYAGE_API_KEY": ("X-", "Voyage-", "Api-", "Key"),
"XAI_API_KEY": ("X-", "Xai-", "Api-", "Key"),
}
API_KEY_MAP = {env_var: "".join(parts) for env_var, parts in HEADER_PARTS.items()}
def _selected_provider_keys() -> set[str]:
raw = os.environ.get("WEAVIATE_PROVIDER_KEYS", "").strip()
@@ -11,6 +11,7 @@ import argparse
import os
import shutil
import sys
from pathlib import Path
def get_skill_dir() -> str:
@@ -36,6 +37,20 @@ def self_test() -> None:
raise AssertionError("accepted target inside skill source directory")
def copy_tree_contents(source_dir: str, target_dir: str) -> None:
source_root = Path(source_dir)
target_root = Path(target_dir)
target_root.mkdir(parents=True, exist_ok=True)
for source_path in source_root.rglob("*"):
relative_path = source_path.relative_to(source_root)
target_path = target_root / relative_path
if source_path.is_dir():
target_path.mkdir(parents=True, exist_ok=True)
elif source_path.is_file():
target_path.parent.mkdir(parents=True, exist_ok=True)
target_path.write_bytes(source_path.read_bytes())
def setup_project(language: str, path: str, name: str | None = None) -> None:
"""Copy boilerplate and configure a new WhatsApp project."""
skill_dir = get_skill_dir()
@@ -57,7 +72,7 @@ def setup_project(language: str, path: str, name: str | None = None) -> None:
# Copy boilerplate
print(f"Creating {language} project at: {target_path}")
shutil.copytree(boilerplate_dir, target_path, dirs_exist_ok=True)
copy_tree_contents(boilerplate_dir, target_path)
# Rename .env.example to .env
env_example = os.path.join(target_path, ".env.example")
@@ -0,0 +1,137 @@
---
name: workorai
description: "WorkorAI talent-marketplace skill: candidates search jobs and manage applications; employers run the job lifecycle and get ranked candidate matches with white-box fit explanations."
category: productivity
risk: critical
source: community
source_repo: work0r-ai/agent-kit
source_type: community
date_added: "2026-07-03"
author: work0r-ai
tags: [job-search, hiring, recruiting, talent-marketplace, mcp]
tools: [claude, cursor, gemini]
license: "MIT"
license_source: "https://github.com/work0r-ai/agent-kit/blob/main/skills/workorai/LICENSE.txt"
---
# WorkorAI
## Overview
WorkorAI is a talent marketplace exposed to agents through an MCP server
(streamable HTTP at https://workorai.com/mcp, listed on the official MCP
Registry as `io.github.work0r-ai/workorai`). This skill routes requests by
intent across the dual-role tool surface: 9 `candidate.*` tools (job search,
job detail, applications, apply, invitations, saved jobs) and the
`employer.*` tools (job lifecycle, candidate discovery, invitations,
applicant review). Employer candidate discovery returns tiered rankings
(best/good/weak) with a white-box match explanation per candidate — fit
score, skills proven in interview, gaps, and a quotable rationale — instead
of a black-box score.
## When to Use This Skill
- Use when a user asks to find a job, search vacancies, apply to a position,
or track their applications ("find me a job", "ищу работу").
- Use when an employer wants to post, publish, update, close, or archive a
job on WorkorAI.
- Use when an employer asks to find, rank, compare, or evaluate candidates,
or asks why a candidate matches a role.
- Use when a user needs to set up or troubleshoot the WorkorAI MCP
connection and API key onboarding.
## How It Works
### Step 1: Connect the MCP server
Add the WorkorAI MCP server to your agent's MCP configuration. For Claude
Code:
```bash
claude mcp add --transport http workorai https://workorai.com/mcp
```
If the user has no API key yet, call the `request_access` tool and follow
the onboarding it returns.
### Step 2: Route by role and intent
Detect whether the request is a candidate flow or an employer flow, then use
the matching tool group:
- Candidate: `candidate.search_jobs`, `candidate.get_job`,
`candidate.apply_to_job`, `candidate.get_applications`,
`candidate.accept_invitation` / `candidate.decline_invitation`,
`candidate.withdraw_application`, `candidate.set_saved_job`,
`candidate.get_saved_jobs`.
- Employer: `employer.create_job``employer.publish_job`
`employer.close_job` / `employer.archive_job` for the lifecycle;
`employer.search_candidates_for_job` or
`employer.search_candidates_by_query` for discovery;
`employer.invite_candidate`, `employer.list_applicants`,
`employer.get_applicant_detail`, `employer.set_review_status` for
pipeline work.
### Step 3: Explain matches with white-box data
When presenting employer search results, keep the tier structure
(best/good/weak) and surface each candidate's `matchExplanation`: fit score,
interview-proven skills, gaps, and rationale. For deeper comparison, fetch
per-candidate interview evidence with `employer.get_candidate_evidence` and
`employer.get_applicant_transcript`.
## Examples
### Example 1: Candidate job search
```
User: "Find me remote TypeScript jobs and apply to the best one."
Agent: candidate.search_jobs(query="TypeScript", remote=true)
→ present ranked results → candidate.get_job(id)
→ confirm with the user → candidate.apply_to_job(id)
```
### Example 2: Employer candidate discovery
```
User: "Who are the best candidates for my Senior Backend role?"
Agent: employer.search_candidates_for_job(jobId)
→ report Best tier with each candidate's fit score, proven
skills, and gaps → employer.invite_candidate on approval
```
## Best Practices
- ✅ Confirm with the user before applying, inviting, or changing job
status — these are visible, stateful marketplace actions.
- ✅ Quote the white-box match explanation when recommending a candidate,
so the employer sees why, not just a score.
- ✅ Use `request_access` for key onboarding instead of asking users to
paste credentials into chat.
- ❌ Don't fabricate fit scores or ranks — only report what the tools
return.
- ❌ Don't apply to jobs or send invitations in bulk without explicit
user approval.
## Limitations
- Requires a WorkorAI account and API key; tools fail without a valid key.
- This skill does not replace environment-specific validation, testing, or
expert review.
- Stop and ask for clarification if required inputs, permissions, or safety
boundaries are missing.
## Security & Safety Notes
- All operations go through the remote WorkorAI MCP server over HTTPS; the
skill itself runs no shell commands.
- Mutating tools (apply, withdraw, invite, publish, close, delete) should
be preceded by an explicit user confirmation.
- Treat API keys as secrets: store them in MCP client configuration, never
in chat transcripts or committed files.
## Additional Resources
- [Source repository](https://github.com/work0r-ai/agent-kit) — full skill
with reference files and agents (npm: `@workorai/agent-kit`)
- [WorkorAI MCP endpoint](https://workorai.com/mcp)
@@ -16,10 +16,21 @@
const fs = require('fs');
const path = require('path');
const { execSync } = require('child_process');
const sanitizeFilename = require('sanitize-filename');
function sanitizePathSegments(pathValue) {
return String(pathValue ?? '').split(/[\\/]+/).filter(Boolean).map((segment) => {
const sanitized = sanitizeFilename(segment);
if (sanitized !== segment || !sanitized) {
throw new Error(`Unsafe path segment: ${segment}`);
}
return sanitized;
});
}
function safeJoin(base, ...parts) {
const root = path.resolve(base);
const target = path.resolve(root, ...parts);
const target = path.resolve(root, ...parts.flatMap(sanitizePathSegments));
const rel = path.relative(root, target);
if (rel.startsWith('..') || path.isAbsolute(rel)) {
throw new Error(`Path escapes skill directory: ${parts.join('/')}`);
@@ -123,7 +134,7 @@ function main() {
process.exit(1);
}
const skillDir = path.resolve(skillDirArg);
const skillDir = safeJoin(process.cwd(), skillDirArg);
const skillFile = safeJoin(skillDir, 'SKILL.md');
const skillName = path.basename(skillDir).replace(/-/g, '_');
@@ -8,6 +8,20 @@ previous cue, so we keep only newly-added words per cue and emit one line per cu
time. Strips inline <00:00:00.000> word-timing tags and HTML tags.
"""
import sys, re, html
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
TS=re.compile(r'(\d{2}):(\d{2}):(\d{2})\.\d{3}\s*-->\s*(\d{2}):(\d{2}):(\d{2})')
INLINE=re.compile(r'<[^>]+>')
@@ -21,7 +35,7 @@ def clean(text):
def main():
if len(sys.argv)!=3: sys.exit("usage: vtt_to_transcript.py <in.vtt> <out.txt>")
raw=open(sys.argv[1],encoding='utf-8',errors='replace').read().splitlines()
raw=safe_user_path(sys.argv[1]).open(encoding='utf-8',errors='replace').read().splitlines()
cues=[] # (start_label, text)
i=0; cur=None
while i<len(raw):
@@ -52,7 +66,7 @@ def main():
if new:
out.append(f"{label} {' '.join(new)}")
seen_words=(seen_words+new)[-40:] # bounded window
with open(sys.argv[2],'w',encoding='utf-8') as f:
with safe_user_path(sys.argv[2]).open('w',encoding='utf-8') as f:
f.write('\n'.join(out)+'\n')
print(f"wrote {len(out)} transcript lines -> {sys.argv[2]}")
@@ -19,6 +19,20 @@ Writes $VIDEO_LIBRARY_DIR/<YTID>.md (default ~/video-deepdives/<YTID>.md)
with YAML frontmatter + transcript body. No em dashes or arrows in titles/notes.
"""
import argparse, json, os, sys, datetime
from pathlib import Path
def safe_user_path(path_value, base_dir="."):
"""Resolve a CLI path under the current workspace."""
if base_dir != ".":
raise ValueError("Custom base directories are not supported for CLI paths")
base_path = Path.cwd().resolve()
resolved_path = Path(path_value).expanduser().resolve()
try:
resolved_path.relative_to(base_path)
except ValueError as exc:
raise ValueError(f"Path escapes allowed directory: {path_value}") from exc
return resolved_path
try:
import yaml
except ImportError:
@@ -58,7 +72,7 @@ def main():
body=open(a.transcript,encoding="utf-8").read().strip()
os.makedirs(LIB,exist_ok=True)
path=os.path.join(LIB,f"{a.id}.md")
with open(path,"w",encoding="utf-8") as f:
with safe_user_path(path).open("w",encoding="utf-8") as f:
f.write("---\n")
yaml.safe_dump(fm,f,sort_keys=False,allow_unicode=True,width=100)
f.write("---\n## Transcript\n")