registry/internal/validators/registries/pypi.go

package registries

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

	"github.com/modelcontextprotocol/registry/pkg/model"
)

var (
	ErrMissingIdentifierForPyPI = errors.New("package identifier is required for PyPI packages")
	ErrMissingVersionForPyPi    = errors.New("package version is required for PyPI packages")
)

// PyPIPackageResponse represents the structure returned by the PyPI JSON API
type PyPIPackageResponse struct {
	Info struct {
		Description string `json:"description"`
	} `json:"info"`
}

// ValidatePyPI validates that a PyPI package contains the correct MCP server name
func ValidatePyPI(ctx context.Context, pkg model.Package, serverName string) error {
	// Set default registry base URL if empty
	if pkg.RegistryBaseURL == "" {
		pkg.RegistryBaseURL = model.RegistryURLPyPI
	}

	if pkg.Identifier == "" {
		return ErrMissingIdentifierForPyPI
	}

	if pkg.Version == "" {
		return ErrMissingVersionForPyPi
	}

	// Validate that MCPB-specific fields are not present
	if pkg.FileSHA256 != "" {
		return fmt.Errorf("PyPI packages must not have 'fileSha256' field - this is only for MCPB packages")
	}

	// Validate that the registry base URL matches PyPI exactly
	if pkg.RegistryBaseURL != model.RegistryURLPyPI {
		return fmt.Errorf("registry type and base URL do not match: '%s' is not valid for registry type '%s'. Expected: %s",
			pkg.RegistryBaseURL, model.RegistryTypePyPI, model.RegistryURLPyPI)
	}

	client := &http.Client{Timeout: 10 * time.Second}

	// PathEscape so an identifier that smuggles "/" or ".." cannot redirect
	// the metadata fetch to a different package than the one being claimed.
	fetchURL := fmt.Sprintf("%s/pypi/%s/%s/json",
		pkg.RegistryBaseURL,
		url.PathEscape(pkg.Identifier),
		url.PathEscape(pkg.Version))
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, fetchURL, nil)
	if err != nil {
		return fmt.Errorf("failed to create request: %w", err)
	}

	req.Header.Set("User-Agent", "MCP-Registry-Validator/1.0")
	req.Header.Set("Accept", "application/json")

	resp, err := client.Do(req)
	if err != nil {
		return fmt.Errorf("failed to fetch package metadata from PyPI: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("PyPI package '%s' not found (status: %d)", pkg.Identifier, resp.StatusCode)
	}

	var pypiResp PyPIPackageResponse
	if err := json.NewDecoder(resp.Body).Decode(&pypiResp); err != nil {
		return fmt.Errorf("failed to parse PyPI package metadata: %w", err)
	}

	// Check description (README) content
	description := pypiResp.Info.Description

	// Check for the mcp-name: <server-name> ownership token (boundary-anchored to
	// avoid prefix confusion — see containsMCPNameToken).
	if containsMCPNameToken(description, serverName) {
		return nil
	}

	// If the token IS present but glued to a trailing character, say so — otherwise
	// the publisher sees "must appear as mcp-name: X" while looking at exactly that.
	if trailing, glued := mcpNameTokenGluedTrailing(description, serverName); glued {
		return fmt.Errorf("PyPI package '%s' ownership validation failed: found 'mcp-name: %s' in the README, but it is immediately followed by %q rather than a boundary. The token must be followed by a space, newline, an HTML tag, or a comment close ('-->') — put it on its own line and republish", pkg.Identifier, serverName, trailing)
	}

	return fmt.Errorf("PyPI package '%s' ownership validation failed. The server name '%s' must appear as 'mcp-name: %s' in the package README", pkg.Identifier, serverName, serverName)
}