wehub-resource-sync
507 lines
17 KiB
Go
507 lines
17 KiB
Go
package github
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
|
|
ghErrors "github.com/github/github-mcp-server/pkg/errors"
|
|
"github.com/github/github-mcp-server/pkg/ifc"
|
|
"github.com/github/github-mcp-server/pkg/inventory"
|
|
"github.com/github/github-mcp-server/pkg/scopes"
|
|
"github.com/github/github-mcp-server/pkg/translations"
|
|
"github.com/github/github-mcp-server/pkg/utils"
|
|
"github.com/google/go-github/v89/github"
|
|
"github.com/google/jsonschema-go/jsonschema"
|
|
"github.com/modelcontextprotocol/go-sdk/mcp"
|
|
)
|
|
|
|
func ListGlobalSecurityAdvisories(t translations.TranslationHelperFunc) inventory.ServerTool {
|
|
return NewTool(
|
|
ToolsetMetadataSecurityAdvisories,
|
|
mcp.Tool{
|
|
Name: "list_global_security_advisories",
|
|
Description: t("TOOL_LIST_GLOBAL_SECURITY_ADVISORIES_DESCRIPTION", "List global security advisories from GitHub."),
|
|
Annotations: &mcp.ToolAnnotations{
|
|
Title: t("TOOL_LIST_GLOBAL_SECURITY_ADVISORIES_USER_TITLE", "List global security advisories"),
|
|
ReadOnlyHint: true,
|
|
},
|
|
InputSchema: &jsonschema.Schema{
|
|
Type: "object",
|
|
Properties: map[string]*jsonschema.Schema{
|
|
"ghsaId": {
|
|
Type: "string",
|
|
Description: "Filter by GitHub Security Advisory ID (format: GHSA-xxxx-xxxx-xxxx).",
|
|
},
|
|
"type": {
|
|
Type: "string",
|
|
Description: "Advisory type.",
|
|
Enum: []any{"reviewed", "malware", "unreviewed"},
|
|
Default: json.RawMessage(`"reviewed"`),
|
|
},
|
|
"cveId": {
|
|
Type: "string",
|
|
Description: "Filter by CVE ID.",
|
|
},
|
|
"ecosystem": {
|
|
Type: "string",
|
|
Description: "Filter by package ecosystem.",
|
|
Enum: []any{"actions", "composer", "erlang", "go", "maven", "npm", "nuget", "other", "pip", "pub", "rubygems", "rust"},
|
|
},
|
|
"severity": {
|
|
Type: "string",
|
|
Description: "Filter by severity.",
|
|
Enum: []any{"unknown", "low", "medium", "high", "critical"},
|
|
},
|
|
"cwes": {
|
|
Type: "array",
|
|
Description: "Filter by Common Weakness Enumeration IDs (e.g. [\"79\", \"284\", \"22\"]).",
|
|
Items: &jsonschema.Schema{
|
|
Type: "string",
|
|
},
|
|
},
|
|
"isWithdrawn": {
|
|
Type: "boolean",
|
|
Description: "Whether to only return withdrawn advisories.",
|
|
},
|
|
"affects": {
|
|
Type: "string",
|
|
Description: "Filter advisories by affected package or version (e.g. \"package1,package2@1.0.0\").",
|
|
},
|
|
"published": {
|
|
Type: "string",
|
|
Description: "Filter by publish date or date range (ISO 8601 date or range).",
|
|
},
|
|
"updated": {
|
|
Type: "string",
|
|
Description: "Filter by update date or date range (ISO 8601 date or range).",
|
|
},
|
|
"modified": {
|
|
Type: "string",
|
|
Description: "Filter by publish or update date or date range (ISO 8601 date or range).",
|
|
},
|
|
},
|
|
},
|
|
},
|
|
[]scopes.Scope{scopes.SecurityEvents},
|
|
func(ctx context.Context, deps ToolDependencies, _ *mcp.CallToolRequest, args map[string]any) (*mcp.CallToolResult, any, error) {
|
|
client, err := deps.GetClient(ctx)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to get GitHub client: %w", err)
|
|
}
|
|
|
|
ghsaID, err := OptionalParam[string](args, "ghsaId")
|
|
if err != nil {
|
|
return utils.NewToolResultError(fmt.Sprintf("invalid ghsaId: %v", err)), nil, nil
|
|
}
|
|
|
|
typ, err := OptionalParam[string](args, "type")
|
|
if err != nil {
|
|
return utils.NewToolResultError(fmt.Sprintf("invalid type: %v", err)), nil, nil
|
|
}
|
|
|
|
cveID, err := OptionalParam[string](args, "cveId")
|
|
if err != nil {
|
|
return utils.NewToolResultError(fmt.Sprintf("invalid cveId: %v", err)), nil, nil
|
|
}
|
|
|
|
eco, err := OptionalParam[string](args, "ecosystem")
|
|
if err != nil {
|
|
return utils.NewToolResultError(fmt.Sprintf("invalid ecosystem: %v", err)), nil, nil
|
|
}
|
|
|
|
sev, err := OptionalParam[string](args, "severity")
|
|
if err != nil {
|
|
return utils.NewToolResultError(fmt.Sprintf("invalid severity: %v", err)), nil, nil
|
|
}
|
|
|
|
cwes, err := OptionalStringArrayParam(args, "cwes")
|
|
if err != nil {
|
|
return utils.NewToolResultError(fmt.Sprintf("invalid cwes: %v", err)), nil, nil
|
|
}
|
|
|
|
isWithdrawn, err := OptionalParam[bool](args, "isWithdrawn")
|
|
if err != nil {
|
|
return utils.NewToolResultError(fmt.Sprintf("invalid isWithdrawn: %v", err)), nil, nil
|
|
}
|
|
|
|
affects, err := OptionalParam[string](args, "affects")
|
|
if err != nil {
|
|
return utils.NewToolResultError(fmt.Sprintf("invalid affects: %v", err)), nil, nil
|
|
}
|
|
|
|
published, err := OptionalParam[string](args, "published")
|
|
if err != nil {
|
|
return utils.NewToolResultError(fmt.Sprintf("invalid published: %v", err)), nil, nil
|
|
}
|
|
|
|
updated, err := OptionalParam[string](args, "updated")
|
|
if err != nil {
|
|
return utils.NewToolResultError(fmt.Sprintf("invalid updated: %v", err)), nil, nil
|
|
}
|
|
|
|
modified, err := OptionalParam[string](args, "modified")
|
|
if err != nil {
|
|
return utils.NewToolResultError(fmt.Sprintf("invalid modified: %v", err)), nil, nil
|
|
}
|
|
|
|
opts := &github.ListGlobalSecurityAdvisoriesOptions{}
|
|
|
|
if ghsaID != "" {
|
|
opts.GHSAID = &ghsaID
|
|
}
|
|
if typ != "" {
|
|
opts.Type = &typ
|
|
}
|
|
if cveID != "" {
|
|
opts.CVEID = &cveID
|
|
}
|
|
if eco != "" {
|
|
opts.Ecosystem = &eco
|
|
}
|
|
if sev != "" {
|
|
opts.Severity = &sev
|
|
}
|
|
if len(cwes) > 0 {
|
|
opts.CWEs = cwes
|
|
}
|
|
|
|
if isWithdrawn {
|
|
opts.IsWithdrawn = &isWithdrawn
|
|
}
|
|
|
|
if affects != "" {
|
|
opts.Affects = &affects
|
|
}
|
|
if published != "" {
|
|
opts.Published = &published
|
|
}
|
|
if updated != "" {
|
|
opts.Updated = &updated
|
|
}
|
|
if modified != "" {
|
|
opts.Modified = &modified
|
|
}
|
|
|
|
advisories, resp, err := client.SecurityAdvisories.ListGlobalSecurityAdvisories(ctx, opts)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to list global security advisories: %w", err)
|
|
}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
body, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to read response body: %w", err)
|
|
}
|
|
return ghErrors.NewGitHubAPIStatusErrorResponse(ctx, "failed to list advisories", resp, body), nil, nil
|
|
}
|
|
|
|
r, err := json.Marshal(advisories)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to marshal advisories: %w", err)
|
|
}
|
|
|
|
result := utils.NewToolResultText(string(r))
|
|
// Global advisories come from the world-readable GitHub Advisory
|
|
// Database (public) but contain externally authored prose
|
|
// (untrusted).
|
|
result = attachStaticIFCLabel(ctx, deps, result, ifc.LabelGlobalSecurityAdvisory())
|
|
return result, nil, nil
|
|
},
|
|
)
|
|
}
|
|
|
|
func ListRepositorySecurityAdvisories(t translations.TranslationHelperFunc) inventory.ServerTool {
|
|
return NewTool(
|
|
ToolsetMetadataSecurityAdvisories,
|
|
mcp.Tool{
|
|
Name: "list_repository_security_advisories",
|
|
Description: t("TOOL_LIST_REPOSITORY_SECURITY_ADVISORIES_DESCRIPTION", "List repository security advisories for a GitHub repository."),
|
|
Annotations: &mcp.ToolAnnotations{
|
|
Title: t("TOOL_LIST_REPOSITORY_SECURITY_ADVISORIES_USER_TITLE", "List repository security advisories"),
|
|
ReadOnlyHint: true,
|
|
},
|
|
InputSchema: &jsonschema.Schema{
|
|
Type: "object",
|
|
Properties: map[string]*jsonschema.Schema{
|
|
"owner": {
|
|
Type: "string",
|
|
Description: "The owner of the repository.",
|
|
},
|
|
"repo": {
|
|
Type: "string",
|
|
Description: "The name of the repository.",
|
|
},
|
|
"direction": {
|
|
Type: "string",
|
|
Description: "Sort direction.",
|
|
Enum: []any{"asc", "desc"},
|
|
},
|
|
"sort": {
|
|
Type: "string",
|
|
Description: "Sort field.",
|
|
Enum: []any{"created", "updated", "published"},
|
|
},
|
|
"state": {
|
|
Type: "string",
|
|
Description: "Filter by advisory state.",
|
|
Enum: []any{"triage", "draft", "published", "closed"},
|
|
},
|
|
},
|
|
Required: []string{"owner", "repo"},
|
|
},
|
|
},
|
|
[]scopes.Scope{scopes.SecurityEvents},
|
|
func(ctx context.Context, deps ToolDependencies, _ *mcp.CallToolRequest, args map[string]any) (*mcp.CallToolResult, any, error) {
|
|
owner, err := RequiredParam[string](args, "owner")
|
|
if err != nil {
|
|
return utils.NewToolResultError(err.Error()), nil, nil
|
|
}
|
|
repo, err := RequiredParam[string](args, "repo")
|
|
if err != nil {
|
|
return utils.NewToolResultError(err.Error()), nil, nil
|
|
}
|
|
|
|
direction, err := OptionalParam[string](args, "direction")
|
|
if err != nil {
|
|
return utils.NewToolResultError(err.Error()), nil, nil
|
|
}
|
|
sortField, err := OptionalParam[string](args, "sort")
|
|
if err != nil {
|
|
return utils.NewToolResultError(err.Error()), nil, nil
|
|
}
|
|
state, err := OptionalParam[string](args, "state")
|
|
if err != nil {
|
|
return utils.NewToolResultError(err.Error()), nil, nil
|
|
}
|
|
|
|
client, err := deps.GetClient(ctx)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to get GitHub client: %w", err)
|
|
}
|
|
|
|
opts := &github.ListRepositorySecurityAdvisoriesOptions{}
|
|
if direction != "" {
|
|
opts.Direction = direction
|
|
}
|
|
if sortField != "" {
|
|
opts.Sort = sortField
|
|
}
|
|
if state != "" {
|
|
opts.State = state
|
|
}
|
|
|
|
advisories, resp, err := client.SecurityAdvisories.ListRepositorySecurityAdvisories(ctx, owner, repo, opts)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to list repository security advisories: %w", err)
|
|
}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
body, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to read response body: %w", err)
|
|
}
|
|
return ghErrors.NewGitHubAPIStatusErrorResponse(ctx, "failed to list repository advisories", resp, body), nil, nil
|
|
}
|
|
|
|
r, err := json.Marshal(advisories)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to marshal advisories: %w", err)
|
|
}
|
|
|
|
result := utils.NewToolResultText(string(r))
|
|
// Repository advisories carry externally authored prose (untrusted).
|
|
// Confidentiality follows repo visibility, but draft/triage/closed
|
|
// advisories are not world-readable even on a public repo, so the
|
|
// result is only public when every returned advisory is published.
|
|
allPublished := allAdvisoriesPublished(advisories)
|
|
result = attachRepoVisibilityIFCLabel(ctx, deps, client, owner, repo, result,
|
|
func(isPrivate bool) ifc.SecurityLabel {
|
|
return ifc.LabelRepositorySecurityAdvisory(isPrivate, allPublished)
|
|
})
|
|
return result, nil, nil
|
|
},
|
|
)
|
|
}
|
|
|
|
func GetGlobalSecurityAdvisory(t translations.TranslationHelperFunc) inventory.ServerTool {
|
|
return NewTool(
|
|
ToolsetMetadataSecurityAdvisories,
|
|
mcp.Tool{
|
|
Name: "get_global_security_advisory",
|
|
Description: t("TOOL_GET_GLOBAL_SECURITY_ADVISORY_DESCRIPTION", "Get a global security advisory"),
|
|
Annotations: &mcp.ToolAnnotations{
|
|
Title: t("TOOL_GET_GLOBAL_SECURITY_ADVISORY_USER_TITLE", "Get a global security advisory"),
|
|
ReadOnlyHint: true,
|
|
},
|
|
InputSchema: &jsonschema.Schema{
|
|
Type: "object",
|
|
Properties: map[string]*jsonschema.Schema{
|
|
"ghsaId": {
|
|
Type: "string",
|
|
Description: "GitHub Security Advisory ID (format: GHSA-xxxx-xxxx-xxxx).",
|
|
},
|
|
},
|
|
Required: []string{"ghsaId"},
|
|
},
|
|
},
|
|
[]scopes.Scope{scopes.SecurityEvents},
|
|
func(ctx context.Context, deps ToolDependencies, _ *mcp.CallToolRequest, args map[string]any) (*mcp.CallToolResult, any, error) {
|
|
client, err := deps.GetClient(ctx)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to get GitHub client: %w", err)
|
|
}
|
|
|
|
ghsaID, err := RequiredParam[string](args, "ghsaId")
|
|
if err != nil {
|
|
return utils.NewToolResultError(fmt.Sprintf("invalid ghsaId: %v", err)), nil, nil
|
|
}
|
|
|
|
advisory, resp, err := client.SecurityAdvisories.GetGlobalSecurityAdvisories(ctx, ghsaID)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to get advisory: %w", err)
|
|
}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
body, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to read response body: %w", err)
|
|
}
|
|
return ghErrors.NewGitHubAPIStatusErrorResponse(ctx, "failed to get advisory", resp, body), nil, nil
|
|
}
|
|
|
|
r, err := json.Marshal(advisory)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to marshal advisory: %w", err)
|
|
}
|
|
|
|
result := utils.NewToolResultText(string(r))
|
|
// A global advisory is world-readable (public) but externally
|
|
// authored (untrusted).
|
|
result = attachStaticIFCLabel(ctx, deps, result, ifc.LabelGlobalSecurityAdvisory())
|
|
return result, nil, nil
|
|
},
|
|
)
|
|
}
|
|
|
|
func ListOrgRepositorySecurityAdvisories(t translations.TranslationHelperFunc) inventory.ServerTool {
|
|
return NewTool(
|
|
ToolsetMetadataSecurityAdvisories,
|
|
mcp.Tool{
|
|
Name: "list_org_repository_security_advisories",
|
|
Description: t("TOOL_LIST_ORG_REPOSITORY_SECURITY_ADVISORIES_DESCRIPTION", "List repository security advisories for a GitHub organization."),
|
|
Annotations: &mcp.ToolAnnotations{
|
|
Title: t("TOOL_LIST_ORG_REPOSITORY_SECURITY_ADVISORIES_USER_TITLE", "List org repository security advisories"),
|
|
ReadOnlyHint: true,
|
|
},
|
|
InputSchema: &jsonschema.Schema{
|
|
Type: "object",
|
|
Properties: map[string]*jsonschema.Schema{
|
|
"org": {
|
|
Type: "string",
|
|
Description: "The organization login.",
|
|
},
|
|
"direction": {
|
|
Type: "string",
|
|
Description: "Sort direction.",
|
|
Enum: []any{"asc", "desc"},
|
|
},
|
|
"sort": {
|
|
Type: "string",
|
|
Description: "Sort field.",
|
|
Enum: []any{"created", "updated", "published"},
|
|
},
|
|
"state": {
|
|
Type: "string",
|
|
Description: "Filter by advisory state.",
|
|
Enum: []any{"triage", "draft", "published", "closed"},
|
|
},
|
|
},
|
|
Required: []string{"org"},
|
|
},
|
|
},
|
|
[]scopes.Scope{scopes.SecurityEvents},
|
|
func(ctx context.Context, deps ToolDependencies, _ *mcp.CallToolRequest, args map[string]any) (*mcp.CallToolResult, any, error) {
|
|
org, err := RequiredParam[string](args, "org")
|
|
if err != nil {
|
|
return utils.NewToolResultError(err.Error()), nil, nil
|
|
}
|
|
direction, err := OptionalParam[string](args, "direction")
|
|
if err != nil {
|
|
return utils.NewToolResultError(err.Error()), nil, nil
|
|
}
|
|
sortField, err := OptionalParam[string](args, "sort")
|
|
if err != nil {
|
|
return utils.NewToolResultError(err.Error()), nil, nil
|
|
}
|
|
state, err := OptionalParam[string](args, "state")
|
|
if err != nil {
|
|
return utils.NewToolResultError(err.Error()), nil, nil
|
|
}
|
|
|
|
client, err := deps.GetClient(ctx)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to get GitHub client: %w", err)
|
|
}
|
|
|
|
opts := &github.ListRepositorySecurityAdvisoriesOptions{}
|
|
if direction != "" {
|
|
opts.Direction = direction
|
|
}
|
|
if sortField != "" {
|
|
opts.Sort = sortField
|
|
}
|
|
if state != "" {
|
|
opts.State = state
|
|
}
|
|
|
|
advisories, resp, err := client.SecurityAdvisories.ListRepositorySecurityAdvisoriesForOrg(ctx, org, opts)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to list organization repository security advisories: %w", err)
|
|
}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
body, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to read response body: %w", err)
|
|
}
|
|
return ghErrors.NewGitHubAPIStatusErrorResponse(ctx, "failed to list organization repository advisories", resp, body), nil, nil
|
|
}
|
|
|
|
r, err := json.Marshal(advisories)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to marshal advisories: %w", err)
|
|
}
|
|
|
|
result := utils.NewToolResultText(string(r))
|
|
// Org-wide advisory listings span the organization's repositories
|
|
// (including private ones) and are restricted to org members, so
|
|
// they are conservatively labeled private-untrusted (isPrivate=true,
|
|
// which forces private regardless of publication state).
|
|
result = attachStaticIFCLabel(ctx, deps, result, ifc.LabelRepositorySecurityAdvisory(true, false))
|
|
return result, nil, nil
|
|
},
|
|
)
|
|
}
|
|
|
|
// allAdvisoriesPublished reports whether every advisory in the slice is in the
|
|
// "published" state. Repository security advisories can also be in draft,
|
|
// triage, or closed states, none of which are world-readable even on a public
|
|
// repository. An empty slice is treated as published (true) since there is no
|
|
// non-public content to protect. Used to decide whether a repository advisory
|
|
// listing may carry a public confidentiality label.
|
|
func allAdvisoriesPublished(advisories []*github.SecurityAdvisory) bool {
|
|
for _, advisory := range advisories {
|
|
if advisory.GetState() != "published" {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|