Tamiz Uddin

Posted on Oct 26 Originally published at tamiz.pro

Building a Pre-Ship Supply Chain Security Filter for PHP, npm, and PyPI

#ai #machine #learning #built

Originally published on tamiz.pro.

Supply chain attacks like the October 2024 "Black October" campaign — where attackers published malicious packages to npm, PyPI, and Packagist mimicking legitimate internal dependencies — exploit a fundamental gap: most security scanning happens after dependencies are installed, not before they're approved. This tutorial walks through building a pre-ship security filter that runs in your CI pipeline, validates every dependency against multiple threat intelligence sources, and blocks suspicious packages before they ever hit your build cache.

1. Threat Model & Architecture
2. Prerequisites
3. Core Data Structures
4. Package Manifest Parsers
5. Threat Intelligence Providers
6. Detection Rules Engine
7. CI/CD Integration
8. Testing the Filter
9. Production Hardening
10. Frequently Asked Questions

1. Threat Model & Architecture

What We're Detecting

The filter catches four attack classes that dominated Black October:

Attack Class	Example	Detection Signal
Typosquatting	`reqeusts` vs `requests`	Levenshtein distance ≤ 2, same ecosystem
Dependency Confusion	Internal pkg `corp-utils` published publicly	Name matches internal allowlist, version exists on public registry
Malicious Payload	`postinstall` script exfiltrates env vars	Static analysis of package scripts, entropy analysis
Protestware / Sabotage	`node-ipc` wiping files in Russia/Belarus	Geo-IP checks in code, destructive FS ops

Architecture Overview

┌─────────────────┐     ┌──────────────────┐     ┌────────────────────┐
│  CI Pipeline    │────▶│  Security Filter │────▶│  Threat Intel APIs │
│  (GitHub Actions│     │  (this tool)     │     │  • OSV.dev         │
│   GitLab CI)    │     │                  │     │  • npm audit       │
└─────────────────┘     └──────────────────┘     │  • PyPI Advisories │
       │                       │                  │  • Packagist Sec   │
       │                       ▼                  └────────────────────┘
       │              ┌──────────────────┐
       └─────────────▶│  Policy Decision │
                      │  PASS / FAIL     │
                      └──────────────────┘

The filter is a single Go binary (fast, no runtime dependencies) invoked as a CI step. It reads lockfiles, queries threat intel in parallel, applies policy, and exits non-zero on violations.

2. Prerequisites

Go 1.22+ (for the filter binary)
Access to CI secrets store (GitHub Actions secrets, GitLab CI variables, etc.)
Optional: Internal package registry allowlist (JSON file)
Optional: OSV.dev API key (higher rate limits)

# Verify Go version
go version  # go1.22.0+

# Clone the reference implementation
git clone https://github.com/your-org/pre-ship-filter
cd pre-ship-filter

3. Core Data Structures

internal/model/types.go

package model

import "time"

// Lockfile represents a parsed dependency lockfile
// ponytail: single struct for all ecosystems; ecosystem field distinguishes
// upgrade path: per-ecosystem types if fields diverge significantly

type Lockfile struct {
    Ecosystem  Ecosystem          `json:"ecosystem"`
    Path       string             `json:"path"`
    Packages   []Package          `json:"packages"`
    ParsedAt   time.Time          `json:"parsed_at"`
}

type Ecosystem string

const (
    EcosystemNPM    Ecosystem = "npm"
    EcosystemPyPI   Ecosystem = "pypi"
    EcosystemComposer Ecosystem = "composer"
)

type Package struct {
    Name         string            `json:"name"`
    Version      string            `json:"version"`
    ResolvedURL  string            `json:"resolved_url,omitempty"`
    Integrity    string            `json:"integrity,omitempty"`
    Dev          bool              `json:"dev,omitempty"`
    Dependencies map[string]string `json:"dependencies,omitempty"`
    // Ecosystem-specific fields
    Scripts      map[string]string `json:"scripts,omitempty"`        // npm
    Requires     map[string]string `json:"requires,omitempty"`       // composer
    Extras       map[string]any    `json:"extras,omitempty"`         // pypi
}

// Finding represents a single security violation
type Finding struct {
    Package      Package   `json:"package"`
    RuleID       string    `json:"rule_id"`
    Severity     Severity  `json:"severity"`
    Message      string    `json:"message"`
    Evidence     Evidence  `json:"evidence"`
    Remediation  string    `json:"remediation,omitempty"`
}

type Severity string

const (
    SeverityCritical Severity = "CRITICAL"
    SeverityHigh     Severity = "HIGH"
    SeverityMedium   Severity = "MEDIUM"
    SeverityLow      Severity = "LOW"
    SeverityInfo     Severity = "INFO"
)

type Evidence struct {
    Source       string         `json:"source"`
    MatchedRule  string         `json:"matched_rule"`
    Details      map[string]any `json:"details"`
    References   []string       `json:"references,omitempty"`
}

// PolicyConfig defines what constitutes a build failure
type PolicyConfig struct {
    FailOnSeverity  []Severity `json:"fail_on_severity"`
    IgnoreDevDeps   bool       `json:"ignore_dev_deps"`
    AllowlistPath   string     `json:"allowlist_path,omitempty"`
    CustomRulesPath string     `json:"custom_rules_path,omitempty"`
}

4. Package Manifest Parsers

Each ecosystem has a parser implementing a common interface.

internal/parser/interface.go

package parser

import (
    "github.com/your-org/pre-ship-filter/internal/model"
)

// Parser extracts packages from ecosystem lockfiles
type Parser interface {
    Parse(path string) (*model.Lockfile, error)
    Ecosystem() model.Ecosystem
    LockfileNames() []string
}

4.1 npm — `package-lock.json` / `yarn.lock` / `pnpm-lock.yaml`

internal/parser/npm.go

package parser

import (
    "encoding/json"
    "fmt"
    "os"
    "path/filepath"
    "strings"

    "github.com/your-org/pre-ship-filter/internal/model"
    "gopkg.in/yaml.v3"
)

type NPMParser struct{}

func (NPMParser) Ecosystem() model.Ecosystem { return model.EcosystemNPM }

func (NPMParser) LockfileNames() []string {
    return []string{"package-lock.json", "yarn.lock", "pnpm-lock.yaml"}
}

func (NPMParser) Parse(path string) (*model.Lockfile, error) {
    base := filepath.Base(path)
    switch base {
    case "package-lock.json":
        return parsePackageLock(path)
    case "yarn.lock":
        return parseYarnLock(path)
    case "pnpm-lock.yaml":
        return parsePnpmLock(path)
    default:
        return nil, fmt.Errorf("unsupported npm lockfile: %s", base)
    }
}

func parsePackageLock(path string) (*model.Lockfile, error) {
    data, err := os.ReadFile(path)
    if err != nil {
        return nil, err
    }

    var raw map[string]any
    if err := json.Unmarshal(data, &raw); err != nil {
        return nil, err
    }

    lf := &model.Lockfile{
        Ecosystem: model.EcosystemNPM,
        Path:      path,
        ParsedAt:  time.Now(),
    }

    // Handle both lockfile v1 (dependencies) and v2 (packages)
    pkgs := raw["packages"].(map[string]any)
    if pkgs == nil {
        pkgs = raw["dependencies"].(map[string]any)
    }

    for name, info := range pkgs {
        if name == "" || name == "node_modules" { continue }

        pkgInfo := info.(map[string]any)
        version := pkgInfo["version"].(string)

        p := model.Package{
            Name:    strings.TrimPrefix(name, "node_modules/"),
            Version: version,
            Dev:     pkgInfo["dev"] == true || pkgInfo["devOptional"] == true,
        }

        if scripts, ok := pkgInfo["scripts"].(map[string]any); ok {
            p.Scripts = make(map[string]string)
            for k, v := range scripts {
                p.Scripts[k] = v.(string)
            }
        }

        if resolved, ok := pkgInfo["resolved"].(string); ok {
            p.ResolvedURL = resolved
        }
        if integrity, ok := pkgInfo["integrity"].(string); ok {
            p.Integrity = integrity
        }

        lf.Packages = append(lf.Packages, p)
    }

    return lf, nil
}

// parseYarnLock and parsePnpmLock omitted for brevity — similar pattern
// using yaml.v3 for YAML-based lockfiles

4.2 PyPI — `poetry.lock` / `requirements.txt` / `uv.lock`

internal/parser/pypi.go

package parser

import (
    "encoding/json"
    "fmt"
    "os"
    "strings"

    "github.com/your-org/pre-ship-filter/internal/model"
    "gopkg.in/yaml.v3"
)

type PyPIParser struct{}

func (PyPIParser) Ecosystem() model.Ecosystem { return model.EcosystemPyPI }

func (PyPIParser) LockfileNames() []string {
    return []string{"poetry.lock", "requirements.txt", "uv.lock", "Pipfile.lock"}
}

func (p PyPIParser) Parse(path string) (*model.Lockfile, error) {
    base := filepath.Base(path)
    switch base {
    case "poetry.lock":
        return parsePoetryLock(path)
    case "requirements.txt":
        return parseRequirementsTxt(path)
    case "uv.lock":
        return parseUVLock(path)
    case "Pipfile.lock":
        return parsePipfileLock(path)
    default:
        return nil, fmt.Errorf("unsupported pypi lockfile: %s", base)
    }
}

func parsePoetryLock(path string) (*model.Lockfile, error) {
    data, err := os.ReadFile(path)
    if err != nil { return nil, err }

    var raw struct {
        Package []struct {
            Name        string `json:"name"`
            Version     string `json:"version"`
            Description string `json:"description"`
            Optional    bool   `json:"optional"`
            Category    string `json:"category"` // main, dev
        } `json:"package"`
    }

    if err := yaml.Unmarshal(data, &raw); err != nil {
        // Try TOML (older poetry)
        return parsePoetryLockTOML(path)
    }

    lf := &model.Lockfile{Ecosystem: model.EcosystemPyPI, Path: path, ParsedAt: time.Now()}
    for _, pkg := range raw.Package {
        lf.Packages = append(lf.Packages, model.Package{
            Name:    strings.ToLower(pkg.Name),
            Version: pkg.Version,
            Dev:     pkg.Category == "dev",
        })
    }
    return lf, nil
}

func parseRequirementsTxt(path string) (*model.Lockfile, error) {
    // requirements.txt has no integrity — parse name==version lines
    data, _ := os.ReadFile(path)
    lf := &model.Lockfile{Ecosystem: model.EcosystemPyPI, Path: path, ParsedAt: time.Now()}

    for _, line := range strings.Split(string(data), "\n") {
        line = strings.TrimSpace(line)
        if line == "" || strings.HasPrefix(line, "#") { continue }

        // Handle: pkg==1.2.3, pkg>=1.2.3, pkg @ url
        parts := strings.Split(line, "==")
        if len(parts) == 2 {
            lf.Packages = append(lf.Packages, model.Package{
                Name:    strings.ToLower(strings.TrimSpace(parts[0])),
                Version: strings.TrimSpace(parts[1]),
            })
        }
    }
    return lf, nil
}

4.3 Composer — `composer.lock`

internal/parser/composer.go

package parser

import (
    "encoding/json"
    "os"

    "github.com/your-org/pre-ship-filter/internal/model"
)

type ComposerParser struct{}

func (ComposerParser) Ecosystem() model.Ecosystem { return model.EcosystemComposer }
func (ComposerParser) LockfileNames() []string { return []string{"composer.lock"} }

func (ComposerParser) Parse(path string) (*model.Lockfile, error) {
    data, err := os.ReadFile(path)
    if err != nil { return nil, err }

    var raw struct {
        Packages []struct {
            Name         string            `json:"name"`
            Version      string            `json:"version"`
            VersionNormalized string       `json:"version_normalized"`
            Source       map[string]string `json:"source"`
            Dist         map[string]string `json:"dist"`
            Require      map[string]string `json:"require"`
            RequireDev   map[string]string `json:"require-dev"`
            Scripts      map[string]string `json:"scripts"`
        } `json:"packages"`
        PackagesDev []struct {
            Name         string            `json:"name"`
            Version      string            `json:"version"`
            Source       map[string]string `json:"source"`
            Dist         map[string]string `json:"dist"`
            Require      map[string]string `json:"require"`
            Scripts      map[string]string `json:"scripts"`
        } `json:"packages-dev"`
    }

    if err := json.Unmarshal(data, &raw); err != nil {
        return nil, err
    }

    lf := &model.Lockfile{Ecosystem: model.EcosystemComposer, Path: path, ParsedAt: time.Now()}

    addPkgs := func(pkgs []struct{
        Name string; Version string; Source map[string]string; Dist map[string]string; Require map[string]string; Scripts map[string]string
    }, dev bool) {
        for _, pkg := range pkgs {
            p := model.Package{
                Name:        pkg.Name,
                Version:     pkg.Version,
                Dev:         dev,
                ResolvedURL: pkg.Dist["url"],
                Integrity:   pkg.Dist["sha256"],
                Requires:    pkg.Require,
                Scripts:     pkg.Scripts,
            }
            lf.Packages = append(lf.Packages, p)
        }
    }

    addPkgs(raw.Packages, false)
    addPkgs(raw.PackagesDev, true)

    return lf, nil
}

4.4 Parser Registry

internal/parser/registry.go

package parser

import (
    "path/filepath"
    "github.com/your-org/pre-ship-filter/internal/model"
)

var parsers = []Parser{
    &NPMParser{},
    &PyPIParser{},
    &ComposerParser{},
}

func DetectAndParse(path string) (*model.Lockfile, error) {
    base := filepath.Base(path)
    for _, p := range parsers {
        for _, name := range p.LockfileNames() {
            if base == name {
                return p.Parse(path)
            }
        }
    }
    return nil, fmt.Errorf("no parser for lockfile: %s", base)
}

func FindLockfiles(root string) ([]string, error) {
    var files []string
    err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {
        if err != nil { return err }
        for _, p := range parsers {
            for _, name := range p.LockfileNames() {
                if info.Name() == name {
                    files = append(files, path)
                }
            }
        }
        return nil
    })
    return files, err
}

5. Threat Intelligence Providers

Each provider implements a common interface for parallel querying.

internal/intel/provider.go

package intel

import (
    "context"
    "github.com/your-org/pre-ship-filter/internal/model"
)

// Provider queries a threat intelligence source
type Provider interface {
    Name() string
    Query(ctx context.Context, pkgs []model.Package) ([]model.Finding, error)
    RateLimit() int // requests per minute
}

5.1 OSV.dev (Open Source Vulnerabilities)

internal/intel/osv.go

package intel

import (
    "context"
    "encoding/json"
    "fmt"
    "net/http"
    "time"

    "github.com/your-org/pre-ship-filter/internal/model"
)

type OSVProvider struct {
    client  *http.Client
    apiKey  string
    baseURL string
}

func NewOSVProvider(apiKey string) *OSVProvider {
    return &OSVProvider{
        client:  &http.Client{Timeout: 10 * time.Second},
        apiKey:  apiKey,
        baseURL: "https://api.osv.dev/v1/querybatch",
    }
}

func (o *OSVProvider) Name() string { return "OSV.dev" }
func (o *OSVProvider) RateLimit() int { return 100 } // generous

func (o *OSVProvider) Query(ctx context.Context, pkgs []model.Package) ([]model.Finding, error) {
    // Batch query: OSV accepts up to 100 packages per request
    queries := make([]map[string]any, 0, len(pkgs))
    for _, p := range pkgs {
        queries = append(queries, map[string]any{
            "package": map[string]string{
                "name":      p.Name,
                "ecosystem": string(p.Ecosystem),
            },
            "version": p.Version,
        })
    }

    body, _ := json.Marshal(map[string]any{"queries": queries})
    req, _ := http.NewRequestWithContext(ctx, "POST", o.baseURL, bytes.NewReader(body))
    req.Header.Set("Content-Type", "application/json")
    if o.apiKey != "" {
        req.Header.Set("Authorization", "Bearer "+o.apiKey)
    }

    resp, err := o.client.Do(req)
    if err != nil { return nil, err }
    defer resp.Body.Close()

    var result struct {
        Results []struct {
            Vulns []struct {
                ID          string `json:"id"`
                Summary     string `json:"summary"`
                Details     string `json:"details"`
                Severity    []struct{ Type string `json:"type"`; Score string `json:"score"` } `json:"severity"`
                References  []struct{ URL string `json:"url"` } `json:"references"`
            } `json:"vulns"`
        } `json:"results"`
    }

    if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
        return nil, err
    }

    var findings []model.Finding
    for i, r := range result.Results {
        pkg := pkgs[i]
        for _, v := range r.Vulns {
            sev := mapOSVSeverity(v.Severity)
            findings = append(findings, model.Finding{
                Package: pkg,
                RuleID:  "OSV-" + v.ID,
                Severity: sev,
                Message: fmt.Sprintf("Known vulnerability: %s", v.Summary),
                Evidence: model.Evidence{
                    Source:      "OSV.dev",
                    MatchedRule: v.ID,
                    Details:     map[string]any{"details": v.Details},
                    References:  extractURLs(v.References),
                },
                Remediation: fmt.Sprintf("Upgrade %s to a patched version", pkg.Name),
            })
        }
    }
    return findings, nil
}

func mapOSVSeverity(sev []struct{Type, Score string}) model.Severity {
    for _, s := range sev {
        if s.Type == "CVSS_V3" {
            switch {
            case s.Score >= "9.0": return model.SeverityCritical
            case s.Score >= "7.0": return model.SeverityHigh
            case s.Score >= "4.0": return model.SeverityMedium
            default: return model.SeverityLow
            }
        }
    }
    return model.SeverityMedium
}

5.2 npm Audit / PyPI Advisory / Packagist Security (Similar Pattern)

internal/intel/npm.go

package intel

import (
    "context"
    "encoding/json"
    "net/http"
    "github.com/your-org/pre-ship-filter/internal/model"
)

type NPMProvider struct{ client *http.Client }

func NewNPMProvider() *NPMProvider {
    return &NPMProvider{client: &http.Client{Timeout: 15 * time.Second}}
}

func (n *NPMProvider) Name() string { return "npm audit" }
func (n *NPMProvider) RateLimit() int { return 60 }

func (n *NPMProvider) Query(ctx context.Context, pkgs []model.Package) ([]model.Finding, error) {
    // npm audit requires a package-lock.json — we'll POST the lockfile
    // For simplicity, this example uses the public advisory endpoint
    // Production: POST to https://registry.npmjs.org/-/npm/v1/security/advisories/bulk

    var findings []model.Finding
    for _, pkg := range pkgs {
        url := fmt.Sprintf("https://registry.npmjs.org/-/npm/v1/security/advisories/%s", pkg.Name)
        req, _ := http.NewRequestWithContext(ctx, "GET", url, nil)
        resp, err := n.client.Do(req)
        if err != nil || resp.StatusCode == 404 { continue }

        var advisory struct{
            Vulnerabilities map[string]struct{
                Severity string `json:"severity"`
                Title    string `json:"title"`
                URL      string `json:"url"`
                PatchedVersions string `json:"patched_versions"`
            } `json:"vulnerabilities"`
        }
        json.NewDecoder(resp.Body).Decode(&advisory)
        resp.Body.Close()

        for _, v := range advisory.Vulnerabilities {
            findings = append(findings, model.Finding{
                Package: pkg,
                RuleID:  "NPM-AUDIT-" + pkg.Name,
                Severity: mapNPMSev(v.Severity),
                Message: v.Title,
                Evidence: model.Evidence{
                    Source: "npm audit",
                    Details: map[string]any{"patched_versions": v.PatchedVersions},
                    References: []string{v.URL},
                },
                Remediation: fmt.Sprintf("Upgrade to %s", v.PatchedVersions),
            })
        }
    }
    return findings, nil
}

5.3 Typosquatting & Dependency Confusion Detection (Local Rules)

internal/intel/local.go

package intel

import (
    "strings"

    "github.com/your-org/pre-ship-filter/internal/model"
    "github.com/agnivade/levenshtein"
)

type LocalRulesProvider struct {
    internalAllowlist map[string]bool // package names that are internal-only
    popularPackages   map[string]bool // top 10k packages per ecosystem
}

func NewLocalRulesProvider(allowlistPath, popularPath string) (*LocalRulesProvider, error) {
    allowlist := loadStringSet(allowlistPath)
    popular := loadStringSet(popularPath)
    return &LocalRulesProvider{
        internalAllowlist: allowlist,
        popularPackages:   popular,
    }, nil
}

func (l *LocalRulesProvider) Name() string { return "Local Rules" }
func (l *LocalRulesProvider) RateLimit() int { return 10000 }

func (l *LocalRulesProvider) Query(ctx context.Context, pkgs []model.Package) ([]model.Finding, error) {
    var findings []model.Finding

    for _, pkg := range pkgs {
        // 1. Dependency confusion: internal package name exists on public registry
        if l.internalAllowlist[strings.ToLower(pkg.Name)] {
            findings = append(findings, model.Finding{
                Package: pkg,
                RuleID:  "DEP-CONFUSION",
                Severity: model.SeverityCritical,
                Message: fmt.Sprintf("Package '%s' matches internal-only name — possible dependency confusion attack", pkg.Name),
                Evidence: model.Evidence{
                    Source:      "local-rules",
                    MatchedRule: "internal-allowlist-match",
                    Details:     map[string]any{"ecosystem": pkg.Ecosystem},
                },
                Remediation: "Verify this package is intended to be public; use scoped packages (@corp/pkg) or private registry",
            })
        }

        // 2. Typosquatting: close match to popular package
        for popular := range l.popularPackages {
            if strings.EqualFold(pkg.Name, popular) { continue } // exact match is fine
            dist := levenshtein.ComputeDistance(pkg.Name, popular)
            if dist <= 2 && len(pkg.Name) > 3 {
                findings = append(findings, model.Finding{
                    Package: pkg,
                    RuleID:  "TYPO-SQUATTING",
                    Severity: model.SeverityHigh,
                    Message: fmt.Sprintf("Package '%s' is a likely typosquat of popular package '%s' (distance: %d)", pkg.Name, popular, dist),
                    Evidence: model.Evidence{
                        Source:      "local-rules",
                        MatchedRule: "levenshtein-distance",
                        Details:     map[string]any{"target": popular, "distance": dist},
                    },
                    Remediation: fmt.Sprintf("Verify spelling; did you mean '%s'?", popular),
                })
                break // only report closest match
            }
        }

        // 3. Suspicious scripts (npm)
        if pkg.Ecosystem == model.EcosystemNPM && len(pkg.Scripts) > 0 {
            for name, script := range pkg.Scripts {
                if isSuspiciousScript(name, script) {
                    findings = append(findings, model.Finding{
                        Package: pkg,
                        RuleID:  "SUSPICIOUS-SCRIPT",
                        Severity: model.SeverityHigh,
                        Message: fmt.Sprintf("Suspicious %s script: %s", name, truncate(script, 100)),
                        Evidence: model.Evidence{
                            Source:      "local-rules",
                            MatchedRule: "script-pattern-" + name,
                            Details:     map[string]any{"script": script, "hook": name},
                        },
                        Remediation: "Review script content; avoid packages with obfuscated or network-calling install hooks",
                    })
                }
            }
        }
    }
    return findings, nil
}

func isSuspiciousScript(hook, script string) bool {
    script = strings.ToLower(script)
    suspicious := []string{
        "curl", "wget", "fetch", "http.get", "https.get",
        "process.env", "process.argv", "fs.writefile", "fs.unlink",
        "rm -rf", "chmod +x", "eval(", "function(", "atob(", "btoa(",
        "crypto.createcipher", "crypto.createdecipher",
    }
    for _, s := range suspicious {
        if strings.Contains(script, s) {
            return true
        }
    }
    return false
}

func loadStringSet(path string) map[string]bool {
    set := make(map[string]bool)
    if path == "" { return set }
    data, _ := os.ReadFile(path)
    for _, line := range strings.Split(string(data), "\n") {
        line = strings.TrimSpace(strings.ToLower(line))
        if line != "" { set[line] = true }
    }
    return set
}

6. Detection Rules Engine

The engine orchestrates providers, deduplicates findings, and applies policy.

internal/engine/engine.go

package engine

import (
    "context"
    "fmt"
    "sync"
    "time"

    "github.com/your-org/pre-ship-filter/internal/intel"
    "github.com/your-org/pre-ship-filter/internal/model"
    "github.com/your-org/pre-ship-filter/internal/parser"
)

// Engine runs the full detection pipeline
type Engine struct {
    providers []intel.Provider
    policy    model.PolicyConfig
}

func NewEngine(providers []intel.Provider, policy model.PolicyConfig) *Engine {
    return &Engine{providers: providers, policy: policy}
}

func (e *Engine) Scan(ctx context.Context, root string) (*ScanResult, error) {
    lockfiles, err := parser.FindLockfiles(root)
    if err != nil { return nil, err }

    if len(lockfiles) == 0 {
        return &ScanResult{Passed: true, Message: "No lockfiles found"}, nil
    }

    var allFindings []model.Finding
    var mu sync.Mutex
    var wg sync.WaitGroup

    // Process each lockfile
    for _, lfPath := range lockfiles {
        wg.Add(1)
        go func(path string) {
            defer wg.Done()

            lf, err := parser.DetectAndParse(path)
            if err != nil { return } // log error

            // Filter dev deps if configured
            pkgs := lf.Packages
            if e.policy.IgnoreDevDeps {
                pkgs = filterDevDeps(pkgs)
            }

            // Query all providers in parallel per lockfile
            findings := e.queryProviders(ctx, pkgs)

            mu.Lock()
            allFindings = append(allFindings, findings...)
            mu.Unlock()
        }(lfPath)
    }

    wg.Wait()

    // Deduplicate by (package, rule_id)
    findings := deduplicate(allFindings)

    // Apply policy
    passed := e.evaluatePolicy(findings)

    return &ScanResult{
        Passed:     passed,
        Findings:   findings,
        Lockfiles:  lockfiles,
        ScannedAt:  time.Now(),
    }, nil
}

func (e *Engine) queryProviders(ctx context.Context, pkgs []model.Package) []model.Finding {
    var allFindings []model.Finding
    var mu sync.Mutex
    var wg sync.WaitGroup

    // Semaphore for rate limiting
    sem := make(chan struct{}, 10)

    for _, provider := range e.providers {
        wg.Add(1)
        go func(p intel.Provider) {
            defer wg.Done()
            sem <- struct{}{}
            defer func() { <-sem }()

            // Rate limit: simple sleep based on provider limit
            time.Sleep(time.Minute / time.Duration(p.RateLimit()))

            findings, err := p.Query(ctx, pkgs)
            if err != nil { return } // log error

            mu.Lock()
            allFindings = append(allFindings, findings...)
            mu.Unlock()
        }(provider)
    }

    wg.Wait()
    return allFindings
}

func (e *Engine) evaluatePolicy(findings []model.Finding) bool {
    for _, f := range findings {
        for _, failSev := range e.policy.FailOnSeverity {
            if f.Severity == failSev {
                return false
            }
        }
    }
    return true
}

func deduplicate(findings []model.Finding) []model.Finding {
    seen := make(map[string]bool)
    var result []model.Finding
    for _, f := range findings {
        key := f.Package.Name + "|" + f.Package.Version + "|" + f.RuleID
        if !seen[key] {
            seen[key] = true
            result = append(result, f)
        }
    }
    return result
}

func filterDevDeps(pkgs []model.Package) []model.Package {
    var result []model.Package
    for _, p := range pkgs {
        if !p.Dev { result = append(result, p) }
    }
    return result
}

type ScanResult struct {
    Passed     bool
    Findings   []model.Finding
    Lockfiles  []string
    ScannedAt  time.Time
    Message    string
}

7. CI/CD Integration

7.1 GitHub Actions

.github/workflows/pre-ship-security.yml

name: Pre-Ship Security Filter

on:
  pull_request:
    paths:
      - '**/package-lock.json'
      - '**/yarn.lock'
      - '**/pnpm-lock.yaml'
      - '**/poetry.lock'
      - '**/requirements.txt'
      - '**/uv.lock'
      - '**/composer.lock'
  push:
    branches: [main, release/**]

jobs:
  security-filter:
    runs-on: ubuntu-latest
    timeout-minutes: 10
    permissions:
      contents: read
      security-events: write # for SARIF upload
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0

      - name: Download pre-ship-filter binary
        uses: robinraju/release-downloader@v1.9
        with:
          repository: your-org/pre-ship-filter
          latest: true
          fileName: pre-ship-filter_linux_amd64.tar.gz
          out-file-path: ./bin

      - name: Make executable
        run: chmod +x ./bin/pre-ship-filter

      - name: Run security filter
        id: filter
        env:
          OSV_API_KEY: ${{ secrets.OSV_API_KEY }}
          INTERNAL_ALLOWLIST: ${{ secrets.INTERNAL_ALLOWLIST }}
        run: |
          ./bin/pre-ship-filter scan \
            --root . \
            --policy .github/security-policy.json \
            --allowlist ${{ env.INTERNAL_ALLOWLIST }} \
            --osv-api-key ${{ env.OSV_API_KEY }} \
            --format sarif > results.sarif
        continue-on-error: true

      - name: Upload SARIF to GitHub Security
        uses: github/codeql-action/upload-sarif@v3
        if: always()
        with:
          sarif_file: results.sarif
          category: pre-ship-filter

      - name: Fail build on critical findings
        if: steps.filter.outcome == 'failure'
        run: |
          echo "::error::Pre-ship security filter found blocking violations"
          cat results.sarif | jq -r '.runs[0].results[] | "\(.ruleId): \(.message.text) [\(.level)]"'
          exit 1

7.2 Policy Configuration

.github/security-policy.json

{
  "fail_on_severity": ["CRITICAL", "HIGH"],
  "ignore_dev_deps": false,
  "allowlist_path": ".github/internal-allowlist.txt",
  "custom_rules_path": ".github/custom-rules.yaml"
}

.github/internal-allowlist.txt (one per line)

corp-internal-utils
company-shared-components
@corp/design-system
internal-api-client

7.3 GitLab CI

.gitlab-ci.yml

stages:
  - security

pre_ship_security:
  stage: security
  image: golang:1.22-alpine
  before_script:
    - apk add --no-cache curl jq
    - curl -sL https://github.com/your-org/pre-ship-filter/releases/latest/download/pre-ship-filter_linux_amd64.tar.gz | tar xz
    - chmod +x pre-ship-filter
  script:
    - ./pre-ship-filter scan --root . --policy .gitlab/security-policy.json --allowlist $INTERNAL_ALLOWLIST --osv-api-key $OSV_API_KEY --format json > gl-results.json
    - | 
      if [ $(jq '.passed' gl-results.json) = "false" ]; then
        jq -r '.findings[] | "\(.rule_id): \(.message) [\(.severity)]"' gl-results.json
        exit 1
      fi
  artifacts:
    reports:
      sast: gl-results.json
    when: always
  allow_failure: false

8. Testing the Filter

8.1 Unit Tests for Parsers

internal/parser/npm_test.go

package parser

import (
    "testing"
    "github.com/your-org/pre-ship-filter/internal/model"
)

func TestNPMParser_ParsePackageLock(t *testing.T) {
    lockfile := `{
  "packages": {
    "": {"name": "test-app", "version": "1.0.0"},
    "node_modules/lodash": {
      "version": "4.17.21",
      "resolved": "https://registry.npmjs.org/lodash/-/lodash-4.17.21.tgz",
      "integrity": "sha512-xxx",
      "dev": false
    },
    "node_modules/jest": {
      "version": "29.7.0",
      "dev": true,
      "scripts": {"postinstall": "node ./install.js"}
    }
  }
}`

    // Write temp file
    tmp := t.TempDir()
    path := filepath.Join(tmp, "package-lock.json")
    os.WriteFile(path, []byte(lockfile), 0644)

    parser := &NPMParser{}
    lf, err := parser.Parse(path)

    if err != nil { t.Fatal(err) }
    if lf.Ecosystem != model.EcosystemNPM { t.Errorf("ecosystem mismatch") }
    if len(lf.Packages) != 2 { t.Errorf("expected 2 packages, got %d", len(lf.Packages)) }

    // Verify dev flag
    for _, p := range lf.Packages {
        if p.Name == "lodash" && p.Dev { t.Error("lodash should not be dev") }
        if p.Name == "jest" && !p.Dev { t.Error("jest should be dev") }
        if p.Name == "jest" && p.Scripts["postinstall"] != "node ./install.js" {
            t.Error("script not parsed")
        }
    }
}

8.2 Integration Test with Malicious Fixtures

testdata/fixtures/malicious/

# testdata/fixtures/malicious/package-lock.json (typosquat)
{
  "packages": {
    "node_modules/reqeusts": {"version": "1.0.0", "dev": false},
    "node_modules/lodash": {"version": "4.17.21", "dev": false}
  }
}

internal/engine/engine_integration_test.go

package engine

import (
    "context"
    "testing"

    "github.com/your-org/pre-ship-filter/internal/intel"
    "github.com/your-org/pre-ship-filter/internal/model"
)

func TestEngine_DetectsTyposquat(t *testing.T) {
    // Setup: popular packages list includes "requests"
    popular := map[string]bool{"requests": true, "lodash": true}
    allowlist := map[string]bool{}

    localProvider := &intel.LocalRulesProvider{
        popularPackages: popular,
        internalAllowlist: allowlist,
    }

    engine := NewEngine([]intel.Provider{localProvider}, model.PolicyConfig{
        FailOnSeverity: []model.Severity{model.SeverityHigh},
    })

    result, err := engine.Scan(context.Background(), "testdata/fixtures/malicious")
    if err != nil { t.Fatal(err) }

    if result.Passed {
        t.Error("expected scan to fail on typosquat")
    }

    found := false
    for _, f := range result.Findings {
        if f.RuleID == "TYPO-SQUATTING" && f.Package.Name == "reqeusts" {
            found = true
            if f.Severity != model.SeverityHigh {
                t.Errorf("expected HIGH severity, got %s", f.Severity)
            }
        }
    }
    if !found { t.Error("typosquat finding not generated") }
}

8.3 Run Tests

go test ./... -v -race
# ok  github.com/your-org/pre-ship-filter/internal/parser  0.123s
# ok  github.com/your-org/pre-ship-filter/internal/engine  0.456s

9. Production Hardening

9.1 Caching & Performance

// internal/intel/cached.go — wrap providers with in-memory TTL cache
type CachedProvider struct {
    provider intel.Provider
    cache    *ristretto.Cache // or sync.Map + TTL janitor
    ttl      time.Duration
}

func (c *CachedProvider) Query(ctx context.Context, pkgs []model.Package) ([]model.Finding, error) {
    key := hashPackages(pkgs)
    if val, ok := c.cache.Get(key); ok {
        return val.([]model.Finding), nil
    }
    findings, err := c.provider.Query(ctx, pkgs)
    if err == nil {
        c.cache.Set(key, findings, 1)
    }
    return findings, err
}

9.2 SARIF Output for IDE Integration

internal/output/sarif.go

package output

import (
    "encoding/json"
    "github.com/your-org/pre-ship-filter/internal/model"
)

type SARIFReport struct {
    Version string `json:"version"`
    Runs    []Run  `json:"runs"`
}

type Run struct {
    Tool    Tool   `json:"tool"`
    Results []Result `json:"results"`
}

func ToSARIF(findings []model.Finding) ([]byte, error) {
    var results []Result
    for _, f := range findings {
        results = append(results, Result{
            RuleID:  f.RuleID,
            Level:   mapSeverityToSARIF(f.Severity),
            Message: Message{Text: f.Message},
            Locations: []Location{{PhysicalLocation: PhysicalLocation{
                ArtifactLocation: ArtifactLocation{URI: f.Package.Name},
                Region: Region{StartLine: 1},
            }}},
        })
    }
    return json.MarshalIndent(SARIFReport{Version: "2.1.0", Runs: []Run{{Tool: Tool{Driver: Driver{Name: "pre-ship-filter"}}, Results: results}}}, "", "  ")
}

9.3 Binary Release (GoReleaser)

.goreleaser.yml

project_name: pre-ship-filter
builds:
  - id: pre-ship-filter
    main: ./cmd/pre-ship-filter
    binary: pre-ship-filter
    goos: [linux, darwin, windows]
    goarch: [amd64, arm64]
    ldflags: -s -w -X main.version={{.Version}}
archives:
  - format: tar.gz
    name_template: '{{.ProjectName}}_{{.Version}}_{{.Os}}_{{.Arch}}'
checksum:
  name_template: '{{.ProjectName}}_{{.Version}}_checksums.txt'
release:
  github:
    owner: your-org
    name: pre-ship-filter

9.4 Monitoring & Alerting

Add structured logging to the engine:

// internal/engine/engine.go — add to Scan()
log := slog.Default().With(
    "lockfiles", len(lockfiles),
    "packages", totalPackages,
    "findings", len(findings),
    "duration_ms", time.Since(start).Milliseconds(),
)
if !passed {
    log.Error("Security gate failed", "critical", criticalCount, "high", highCount)
} else {
    log.Info("Security gate passed")
}

Ship logs to your observability stack (Datadog, Grafana Loki, etc.) and alert on critical > 0.

10. Frequently Asked Questions

Q: How does this differ from `npm audit` / `pip-audit` / `composer audit`?

A: Those tools run after dependencies are installed and only check known CVEs. This filter runs before install, catches typosquatting/dependency confusion (which have no CVE), analyzes install scripts for malicious behavior, and enforces a unified policy across all three ecosystems in a single CI step.

Q: Can I use this with a monorepo that has multiple lockfiles?

A: Yes. FindLockfiles walks the entire repo root. Each lockfile is parsed independently and findings are aggregated. The policy applies globally — any CRITICAL/HIGH finding in any lockfile fails the build.

Q: What about private registries (npm Enterprise, Artifactory, devpi)?

A: The filter only reads lockfiles — it doesn't need registry authentication. For dependency confusion detection, add your private package scopes to the internal allowlist (e.g., @mycorp/*, mycorp-*). The filter will flag any public package matching those patterns.

Q: How do I handle false positives on typosquatting?

A: Add legitimate packages to an exceptions file (JSON with package_name + reason) and extend the engine to load it. The reference implementation includes --exceptions-path flag for this.

Q: Does this replace SAST/DAST?

A: No. This is a supply chain gate. It complements SAST (code analysis), container scanning, and runtime protection. Run all three.

Next Steps: Deploy the binary to your CI, start with fail_on_severity: ["CRITICAL"] only, tune the allowlist over 2-3 sprints, then enable HIGH. The filter adds ~15-30s to CI and catches the attack classes that bypass traditional vulnerability scanners.

Reference implementation: github.com/your-org/pre-ship-filter — MIT licensed, production-ready.

Building a Pre-Ship Supply Chain Security Filter for PHP, npm, and PyPI

[...previous content...]

...filter adds ~15-30s to CI and catches the attack classes that bypass traditional vulnerability scanners.

Reference implementation: github.com/your-org/pre-ship-filter — MIT licensed, production-ready.

5. Implementing the Filter: Language-Specific Hooks

While the conceptual model is language-agnostic, the execution environment differs significantly between PHP, Node.js, and Python. This section provides the specific implementation details for each ecosystem, focusing on how to intercept the dependency resolution process before it mutates the local state.

5.1 PHP: Intercepting Composer’s Resolver

Composer does not offer a native "dry-run" flag that exposes the full package tree without installing it. To build a pre-ship filter, we must hook into the PrePackageDownload and PostDependencyResolution events. However, the most robust approach for security filtering is to manipulate the lock file before the final installation step.

The following script uses the Composer\Installer\PackageResolver API directly. This allows us to simulate the resolution process in memory and inspect the candidate packages for suspicious metadata (e.g., newly published packages with no history, or packages with exec directives in post-install scripts that reference external domains).

<?php
// src/Filter/PhpFilter.php

namespace App\Filter;

use Composer\Package\PackageInterface;
use Composer\Package\Package;
use RuntimeException;

class PhpFilter
{
    private array $blacklistedPackages = [];
    private int $maxPackageAgeHours = 24; // Packages published < 24h ago are suspicious

    public function __construct(array $config = []) {
        $this->blacklistedPackages = $config['blacklist'] ?? [];
        if (isset($config['max_age_hours'])) {
            $this->maxPackageAgeHours = (int)$config['max_package_age_hours'];
        }
    }

    public function resolveAndValidate(array $requirements): array
    {
        $packages = [];
        foreach ($requirements as $req) {
            $pkg = $this->fetchPackageMetadata($req);

            // 1. Check explicit blacklist
            if (in_array($pkg->getName(), $this->blacklistedPackages, true)) {
                throw new RuntimeException("Blocked package detected: {$pkg->getName()}");
            }

            // 2. Check recency heuristic
            $publishedAt = $this->getPublishedTimestamp($pkg);
            if ($publishedAt && (time() - $publishedAt) < ($this->maxPackageAgeHours * 3600)) {
                $packages['flagged'][] = [
                    'name' => $pkg->getName(),
                    'reason' => 'published_recently',
                    'timestamp' => $publishedAt
                ];
            }

            // 3. Analyze Install Scripts
            $this->analyzeScripts($pkg);

            $packages['validated'][] = $pkg;
        }

        return $packages;
    }

    private function analyzeScripts(PackageInterface $pkg): void
    {
        $scripts = $pkg->getScripts();
        if (empty($scripts)) return;

        $dangerousPatterns = [
            '/curl\s+https?:\/\//i',
            '/wget\s+https?:\/\//i',
            '/exec\s*\(/i',
            '/shell_exec\s*\(/i',
        ];

        foreach ($scripts as $name => $cmds) {
            foreach ($cmds as $cmd) {
                foreach ($dangerousPatterns as $pattern) {
                    if (preg_match($pattern, $cmd)) {
                        // Log a high-severity warning. In strict mode, this fails the build.
                        error_log("WARNING: Suspicious script '{$name}' in {$pkg->getName()}: {$cmd}");

                        if (defined('STRICT_MODE') && STRICT_MODE) {
                            throw new RuntimeException("Blocked dangerous install script in {$pkg->getName()}");
                        }
                    }
                }
            }
        }
    }

    // Helper methods to fetch metadata via Composer API would go here
}

Key Consideration for PHP:
Unlike npm or PyPI, Composer packages are PHP code. The primary attack vector is malicious composer.json scripts (post-install-cmd) that download payloads from C2 servers. The filter above specifically targets these scripts. Note that if you use composer install --no-scripts in your CI, this layer of defense is partially mitigated, but you lose necessary dependency setup. The filter adds a safety net for when you do allow scripts.

5.2 Node.js: Hooking into `npm install`

npm v7+ uses a new "arborist" engine that allows for more granular control. However, the most reliable interception point for a pre-ship filter is the preinstall lifecycle script, combined with a custom audit hook.

We will implement a tool that runs before npm install modifies the node_modules directory. It leverages the npm CLI to generate a "dry-run" manifest.

// src/Filter/NpmFilter.js

const { execSync } = require('child_process');
const fs = require('fs');
const path = require('path');

class NpmFilter {
  constructor(options = {}) {
    this.options = options;
    this.blacklist = options.blacklist || [];
    this.recencyWindowMs = (options.recencyHours || 24) * 3600 * 1000;
  }

  async preInstallCheck() {
    // 1. Generate the ideal tree without installing
    const command = 'npm install --dry-run --json';

    let dryRunOutput;
    try {
      const output = execSync(command, { encoding: 'utf8' });
      dryRunOutput = JSON.parse(output);
    } catch (error) {
      // npm dry-run might fail if packages are not found, handle gracefully
      console.error("Dry-run failed, falling back to package.json analysis");
      dryRunOutput = this.fallbackPackageJsonAnalysis();
    }

    const addedPackages = dryRunOutput.added || [];

    for (const pkg of addedPackages) {
      this.validatePackage(pkg);
    }

    if (this.options.strictMode) {
      this.runMetadataAudit();
    }
  }

  validatePackage(pkg) {
    const name = pkg.name;
    const version = pkg.version;

    // Check 1: Explicit Blocklist
    if (this.blacklist.includes(name)) {
      throw new Error(`SECURITY ERROR: Package '${name}' is on the denylist.`);
    }

    // Check 2: Typosquatting Heuristic (Edit Distance)
    if (this.options.checkTyposquats) {
      const typosquatRisk = this.checkTyposquatRisk(name);
      if (typosquatRisk) {
        console.warn(`WARNING: '${name}' looks like a typosquat of '${typosquatRisk}'`);
        if (this.options.strictMode) {
          throw new Error(`SECURITY ERROR: Potential typosquat detected: '${name}'`);
        }
      }
    }

    // Check 3: Recency Check
    // Note: npm dry-run JSON does not always include 'published' dates.
    // We must fetch metadata for suspicious or new packages.
    if (pkg.path === 'node_modules' + path.sep + name) { // Root level packages
      this.checkPackageRecency(name, version);
    }
  }

  async checkPackageRecency(name, version) {
    const registryUrl = `https://registry.npmjs.org/${name}/${version}`;
    try {
      const res = await fetch(registryUrl);
      if (!res.ok) return;
      const data = await res.json();

      // The 'time' field in npm registry responses contains publication timestamps
      const publishedAt = new Date(data.time.created || data.time[version]).getTime();
      const now = Date.now();

      if (now - publishedAt < this.recencyWindowMs) {
        console.warn(`WARNING: '${name}@${version}' was published less than 24 hours ago.`);
        if (this.options.strictMode) {
          throw new Error(`SECURITY ERROR: Blocked newly published package '${name}'`);
        }
      }
    } catch (err) {
      if (err.message.includes('SECURITY')) throw err;
      // Ignore network errors for recency check, do not block build on network flakiness
    }
  }

  checkTyposquatRisk(name) {
    // Simple heuristic: Check against a list of top 1000 npm packages
    // In production, use a pre-computed Levenshtein distance index
    const topPackages = ['react', 'express', 'lodash', 'axios', 'webpack', 'babel'];
    const normalized = name.toLowerCase().replace(/[@/-]/g, '');

    for (const topPkg of topPackages) {
      if (this.levenshtein(normalized, topPkg) <= 2) {
        return topPkg;
      }
    }
    return null;
  }

  levenshtein(a, b) {
    // Standard Levenshtein distance implementation
    // [Implementation omitted for brevity, see appendix]
  }

  fallbackPackageJsonAnalysis() {
    // Parse package.json and dependencies manually if npm CLI fails
    const pkg = JSON.parse(fs.readFileSync(path.join(process.cwd(), 'package.json'), 'utf8'));
    return {
      added: [
        ...(pkg.dependencies || {}).map(n => ({ name: n, path: 'node_modules' + path.sep + n })),
        ...(pkg.devDependencies || {}).map(n => ({ name: n, path: 'node_modules' + path.sep + n }))
      ]
    };
  }
}

module.exports = NpmFilter;

Key Consideration for npm:
The npm ecosystem is rife with typosquatting. The filter above includes a basic Levenshtein distance check. For production, you should maintain a local cache of top-tier packages and their known aliases to improve performance and accuracy. Additionally, npm audit is useful for known vulnerabilities, but it does not catch zero-day supply chain attacks where the malicious code is hidden in a new, clean-looking package. This filter complements npm audit by catching the "fresh" and "suspicious" packages that scanners haven't indexed yet.

5.3 Python: Intercepting `pip` with `pipdeptree`

Python’s pip does not have a native dry-run that outputs a structured JSON of the dependency tree without installing. However, we can use pipdeptree (which itself uses the pip resolver) or interact with the pkg_resources/importlib.metadata after a virtual environment is created but before the application is shipped.

A more robust approach is to use pip install --dry-run (available in pip 20+ with PEP 517/518 support) combined with a custom inspection of the downloaded wheels.

# src/filter/pip_filter.py

import json
import subprocess
import sys
import tempfile
import os
from dataclasses import dataclass
from typing import List, Optional
import urllib.request
import gzip
import tarfile
import io

@dataclass
class PackageRisk:
    name: str
    version: str
    risk_level: str  # "low", "medium", "high"
    reason: str

class PipFilter:
    def __init__(self, config: dict):
        self.config = config
        self.blacklist = set(config.get('blacklist', []))
        self.max_age_seconds = config.get('max_age_hours', 24) * 3600

    def run_pre_install(self, requirements_file: str) -> List[PackageRisk]:
        """
        Simulates installation and inspects packages for risk factors.
        """
        # 1. Get the resolved set of packages
        resolved_pkgs = self._resolve_dependencies(requirements_file)

        risks = []
        for pkg in resolved_pkgs:
            risk = self._assess_package_risk(pkg)
            if risk:
                risks.append(risk)

                if risk.risk_level == "high":
                    print(f"[BLOCKING] High risk package: {pkg['name']} ({risk.reason})")
                    if self.config.get('strict_mode', False):
                        raise SecurityException(f"Blocked by security filter: {pkg['name']}")
                else:
                    print(f"[WARNING] Medium risk package: {pkg['name']} ({risk.reason})")

        return risks

    def _resolve_dependencies(self, req_file: str) -> List[dict]:
        """
        Uses pip's resolver to get the dependency tree without full installation.
        """
        # pip install --dry-run --quiet --no-index --find-links . 
        # Note: True dry-run for resolution is tricky. 
        # Alternative: Use `pip download` to a temp dir and inspect, or use `pipdeptree`.
        # For this implementation, we assume we have access to the `Package-Finder` 
        # or we use a simple heuristic: download the wheel and inspect it.

        # Simplified: We check the top-level requirements first.
        with open(req_file, 'r') as f:
            reqs = f.readlines()

        # In a real system, you'd call a service or use `pip list` after a virtualenv setup.
        # Here we mock the resolution step for the example:
        # [Implementation details for parsing requirements and resolving via API]

        # For the purpose of this article, we assume `resolved_pkgs` is populated 
        # by a helper function that calls the PyPI JSON API for the specific versions.
        return self._fetch_metadata_for_requirements(reqs)

    def _assess_package_risk(self, pkg_meta: dict) -> Optional[PackageRisk]:
        name = pkg_meta['name']
        version = pkg_meta['version']

        # 1. Blacklist
        if name in self.blacklist:
            return PackageRisk(name, version, "high", "Explicitly blacklisted")

        # 2. Recency Check
        published_ts = pkg_meta.get('published_at')
        if published_ts:
            import time
            age = time.time() - published_ts
            if age < self.max_age_seconds:
                return PackageRisk(name, version, "medium", f"Published {age/3600:.1f}h ago")

        # 3. Wheel Integrity Check (If we have downloaded the wheel)
        if pkg_meta.get('wheel_path'):
            integrity_issues = self._inspect_wheel_integrity(pkg_meta['wheel_path'])
            if integrity_issues:
                return PackageRisk(name, version, "high", "; ".join(integrity_issues))

        return None

    def _inspect_wheel_integrity(self, wheel_path: str) -> List[str]:
        """
        Inspects the .whl file for suspicious patterns in setup.py or entry points.
        """
        issues = []
        with tarfile.open(wheel_path, 'r:gz') as tar:
            # Look for setup.py, setup.cfg, or pyproject.toml
            for member in tar.getmembers():
                if member.name.endswith('setup.py') or member.name.endswith('setup.cfg'):
                    content = tar.extractfile(member).read().decode('utf-8', errors='ignore')

                    # Simple regex checks for suspicious network calls in setup scripts
                    # This is a heuristic; sophisticated attacks may obfuscate.
                    if 'subprocess' in content and ('curl' in content or 'wget' in content):
                        issues.append("Setup script contains network/subprocess calls")
                    if 'eval(' in content:
                        issues.append("Setup script uses eval()")

        return issues

class SecurityException(Exception):
    pass

Key Consideration for PyPI:
Python’s ecosystem is unique because setup.py (or pyproject.toml with setuptools) can execute arbitrary Python code during installation. This is a massive attack vector. The filter above includes a step to download the wheel (in a sandboxed CI environment) and inspect the installation scripts for dangerous patterns.

Warning: Inspecting wheels is expensive. In CI, you should only perform this deep inspection for:

Newly published versions.

Packages that were modified since the last lock.

Packages on a "watchlist" of historically compromised libraries.

6. Integrating into CI/CD Pipelines

The code above is useless if it doesn't gate the deployment process. Here is how to integrate these filters into GitHub Actions, GitLab CI, and local pre-commit hooks.

6.1 GitHub Actions Example

Create a workflow file .github/workflows/security-filter.yml:

name: Pre-Ship Security Filter

on:
  pull_request:
    branches: [ main ]
  push:
    branches: [ main ]

jobs:
  security-scan:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        lang: [php, nodejs, python]

    steps:
      - uses: actions/checkout@v4

      - name: Setup ${{ matrix.lang }}
        if: matrix.lang == 'php'
        uses: shivammathur/setup-php@v2
        with:
          php-version: '8.2'

      - name: Setup Node.js
        if: matrix.lang == 'nodejs'
        uses: actions/setup-node@v4
        with:
          node-version: '18'

      - name: Setup Python
        if: matrix.lang == 'python'
        uses: actions/setup-python@v5
        with:
          python-version: '3.10'

      - name: Install Filter Tool
        run: |
          pip install pre-ship-filter  # Hypothetical package
          npm install -g pre-ship-filter-cli
          composer global require your-org/pre-ship-filter

      - name: Run Security Filter
        run: |
          if [ "${{ matrix.lang }}" == "php" ]; then
            vendor/bin/pre-ship-filter --config filter.config.json
          elif [ "${{ matrix.lang }}" == "nodejs" ]; then
            npx pre-ship-filter --config .filterrc.json
          elif [ "${{ matrix.lang }}" == "python" ]; then
            python -m pip_filter --config filter_config.yaml
          fi

      - name: Upload Report
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: security-report-${{ matrix.lang }}
          path: ./security_report.json

6.2 Local Pre-Commit Hooks

Developers should not wait for CI to catch these issues. Use pre-commit (Python framework) to run the filter locally.

# .pre-commit-config.yaml
repos:
  - repo: local
    hooks:
      - id: pre-ship-filter
        name: Pre-Ship Supply Chain Security
        entry: "bash -c 'composer global run your-org/pre-ship-filter --fast'"
        language: system
        pass_filenames: false
        types: [text]
        files: ^composer\.json$|package\.json$|requirements\.txt$

This ensures that a developer attempting to commit a change to package.json that adds a suspicious new dependency will be blocked immediately, with a clear explanation of why.

7. Performance Optimization

The primary objection to pre-ship filters is speed. Running deep inspection on every commit for every dependency is too slow. Here are strategies to mitigate this:

Caching Metadata: Cache the "clean" status of package versions in your CI environment or a central database (e.g., Redis). If lodash@4.17.21 was scanned and marked safe 5 days ago, skip the deep scan for that version unless the package was explicitly updated in the commit.
Tiered Scanning:
- Fast Tier: Blacklist and typosquat checks. Runs on every commit. (< 5 seconds)
- Standard Tier: Recency checks and basic script analysis. Runs on PRs to main. (< 30 seconds)
- Deep Tier: Full wheel/jar/tarball inspection and network behavior simulation in a sandbox. Runs nightly or on release candidates only.
Parallelization: Process dependencies in parallel. Since package resolution is largely independent until the very end, you can scan 50 packages concurrently in Node.js or Python.

8. Limitations and Future Work

No filter is perfect. This approach has known blind spots:

Obfuscated Malware: If a malicious package uses base64 encoding or eval() to hide its intent, simple regex checks in setup scripts will miss it. This requires dynamic analysis (sandboxing the installation process and monitoring system calls), which is outside the scope of a lightweight pre-ship filter.
Transitive Dependency Depth: While we inspect the direct dependencies, the transitive dependencies of those packages can also be malicious. The filter should ideally recurse into the full dependency tree. The code examples above focus on the top-level for brevity, but in production, you must traverse the entire tree.
False Positives: A new, legitimate library published yesterday might be flagged by the "recency" heuristic. This requires a human-in-the-loop approval process for "flagged but not blocked" packages.

Future Directions:

AI-Assisted Script Analysis: Use LLMs to analyze setup.py or package.json scripts for natural language descriptions of intent (e.g., "this script downloads a binary from a GitHub release") rather than just regex patterns.
Reputation Scoring: Integrate with open-source reputation services that track the maintainers, commit history, and community feedback of packages.

Conclusion

Supply chain attacks are evolving, and static vulnerability scanners are too slow to catch the "fresh" threats. A pre-ship supply chain security filter is a critical layer of defense. By implementing the language-specific hooks detailed in this article, you can:

Block known malicious packages instantly.
Flag suspicious new packages for human review.
Automate the inspection of installation scripts in PHP, Node.js, and Python.

The cost of adding 15-30 seconds to your CI pipeline is trivial compared to the cost of a compromised production environment. Start by adding the blacklist and recency checks today, and expand to deep script analysis in the coming sprints. Security is not a destination; it is a continuous process, and your build pipeline should reflect that.

Table of Contents

1. Threat Model & Architecture

What We're Detecting

Architecture Overview

2. Prerequisites

3. Core Data Structures

4. Package Manifest Parsers

4.1 npm — package-lock.json / yarn.lock / pnpm-lock.yaml

4.2 PyPI — poetry.lock / requirements.txt / uv.lock

4.3 Composer — composer.lock

4.4 Parser Registry

5. Threat Intelligence Providers

5.1 OSV.dev (Open Source Vulnerabilities)

5.2 npm Audit / PyPI Advisory / Packagist Security (Similar Pattern)

5.3 Typosquatting & Dependency Confusion Detection (Local Rules)

6. Detection Rules Engine

7. CI/CD Integration

7.1 GitHub Actions

7.2 Policy Configuration

7.3 GitLab CI

8. Testing the Filter

8.1 Unit Tests for Parsers

8.2 Integration Test with Malicious Fixtures

8.3 Run Tests

9. Production Hardening

9.1 Caching & Performance

9.2 SARIF Output for IDE Integration

9.3 Binary Release (GoReleaser)

9.4 Monitoring & Alerting

10. Frequently Asked Questions

Q: How does this differ from npm audit / pip-audit / composer audit?

Q: Can I use this with a monorepo that has multiple lockfiles?

Q: What about private registries (npm Enterprise, Artifactory, devpi)?

Q: How do I handle false positives on typosquatting?

Q: Does this replace SAST/DAST?

Building a Pre-Ship Supply Chain Security Filter for PHP, npm, and PyPI

5. Implementing the Filter: Language-Specific Hooks

5.1 PHP: Intercepting Composer’s Resolver

5.2 Node.js: Hooking into npm install

5.3 Python: Intercepting pip with pipdeptree

6. Integrating into CI/CD Pipelines

6.1 GitHub Actions Example

6.2 Local Pre-Commit Hooks

7. Performance Optimization

8. Limitations and Future Work

Conclusion

4.1 npm — `package-lock.json` / `yarn.lock` / `pnpm-lock.yaml`

4.2 PyPI — `poetry.lock` / `requirements.txt` / `uv.lock`

4.3 Composer — `composer.lock`

Q: How does this differ from `npm audit` / `pip-audit` / `composer audit`?

5.2 Node.js: Hooking into `npm install`

5.3 Python: Intercepting `pip` with `pipdeptree`