gitea-mcp/operation/repo/branch_protection.go

package repo

import (
	"context"
	"fmt"

	"gitea.com/gitea/gitea-mcp/pkg/gitea"
	"gitea.com/gitea/gitea-mcp/pkg/log"
	"gitea.com/gitea/gitea-mcp/pkg/params"
	"gitea.com/gitea/gitea-mcp/pkg/to"

	gitea_sdk "code.gitea.io/sdk/gitea"
	"github.com/mark3labs/mcp-go/mcp"
	"github.com/mark3labs/mcp-go/server"
)

const (
	BranchProtectionReadToolName  = "branch_protection_read"
	BranchProtectionWriteToolName = "branch_protection_write"
)

var (
	BranchProtectionReadTool = mcp.NewTool(
		BranchProtectionReadToolName,
		mcp.WithDescription("Read branch protection rules. Use method 'list' to list all rules, 'get' to get a specific rule."),
		mcp.WithString("method", mcp.Required(), mcp.Description("operation to perform"), mcp.Enum("list", "get")),
		mcp.WithString("owner", mcp.Required(), mcp.Description("repository owner")),
		mcp.WithString("repo", mcp.Required(), mcp.Description("repository name")),
		mcp.WithString("name", mcp.Description("branch protection rule name (required for 'get')")),
		mcp.WithNumber("page", mcp.Description("page number"), mcp.DefaultNumber(1)),
		mcp.WithNumber("perPage", mcp.Description("results per page"), mcp.DefaultNumber(30)),
	)

	BranchProtectionWriteTool = mcp.NewTool(
		BranchProtectionWriteToolName,
		mcp.WithDescription("Create, edit, or delete branch protection rules."),
		mcp.WithString("method", mcp.Required(), mcp.Description("operation to perform"), mcp.Enum("create", "edit", "delete")),
		mcp.WithString("owner", mcp.Required(), mcp.Description("repository owner")),
		mcp.WithString("repo", mcp.Required(), mcp.Description("repository name")),
		mcp.WithString("branch_name", mcp.Description("branch name or pattern (required for 'create')")),
		mcp.WithString("rule_name", mcp.Description("rule name (required for 'edit', 'delete')")),
		mcp.WithBoolean("enable_push", mcp.Description("enable push to protected branch")),
		mcp.WithArray("push_whitelist_usernames", mcp.Description("usernames allowed to push"), mcp.Items(map[string]any{"type": "string"})),
		mcp.WithArray("push_whitelist_teams", mcp.Description("team names allowed to push"), mcp.Items(map[string]any{"type": "string"})),
		mcp.WithBoolean("enable_merge_whitelist", mcp.Description("enable merge whitelist")),
		mcp.WithArray("merge_whitelist_usernames", mcp.Description("usernames allowed to merge"), mcp.Items(map[string]any{"type": "string"})),
		mcp.WithArray("merge_whitelist_teams", mcp.Description("team names allowed to merge"), mcp.Items(map[string]any{"type": "string"})),
		mcp.WithNumber("required_approvals", mcp.Description("number of required approvals")),
		mcp.WithBoolean("block_on_rejected_reviews", mcp.Description("block merge on rejected reviews")),
		mcp.WithBoolean("block_on_outdated_branch", mcp.Description("block merge on outdated branch")),
		mcp.WithBoolean("dismiss_stale_approvals", mcp.Description("dismiss stale approvals on new commits")),
		mcp.WithBoolean("require_signed_commits", mcp.Description("require signed commits")),
		mcp.WithString("protected_file_patterns", mcp.Description("protected file patterns (semicolon separated)")),
		mcp.WithString("unprotected_file_patterns", mcp.Description("unprotected file patterns (semicolon separated)")),
	)
)

func init() {
	Tool.RegisterRead(server.ServerTool{
		Tool:    BranchProtectionReadTool,
		Handler: branchProtectionReadFn,
	})
	Tool.RegisterWrite(server.ServerTool{
		Tool:    BranchProtectionWriteTool,
		Handler: branchProtectionWriteFn,
	})
}

func branchProtectionReadFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	args := req.GetArguments()
	method, err := params.GetString(args, "method")
	if err != nil {
		return to.ErrorResult(err)
	}
	switch method {
	case "list":
		return listBranchProtectionsFn(ctx, req)
	case "get":
		return getBranchProtectionFn(ctx, req)
	default:
		return to.ErrorResult(fmt.Errorf("unknown method: %s", method))
	}
}

func listBranchProtectionsFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	log.Debugf("Called listBranchProtectionsFn")
	args := req.GetArguments()
	owner, err := params.GetString(args, "owner")
	if err != nil {
		return to.ErrorResult(err)
	}
	repo, err := params.GetString(args, "repo")
	if err != nil {
		return to.ErrorResult(err)
	}
	client, err := gitea.ClientFromContext(ctx)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("get gitea client err: %v", err))
	}
	bps, _, err := client.ListBranchProtections(owner, repo, gitea_sdk.ListBranchProtectionsOptions{})
	if err != nil {
		return to.ErrorResult(fmt.Errorf("list branch protections %v/%v err: %v", owner, repo, err))
	}
	return to.TextResult(slimBranchProtections(bps))
}

func getBranchProtectionFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	log.Debugf("Called getBranchProtectionFn")
	args := req.GetArguments()
	owner, err := params.GetString(args, "owner")
	if err != nil {
		return to.ErrorResult(err)
	}
	repo, err := params.GetString(args, "repo")
	if err != nil {
		return to.ErrorResult(err)
	}
	name, err := params.GetString(args, "name")
	if err != nil {
		return to.ErrorResult(err)
	}
	client, err := gitea.ClientFromContext(ctx)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("get gitea client err: %v", err))
	}
	bp, _, err := client.GetBranchProtection(owner, repo, name)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("get branch protection %v in %v/%v err: %v", name, owner, repo, err))
	}
	return to.TextResult(slimBranchProtection(bp))
}

func branchProtectionWriteFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	args := req.GetArguments()
	method, err := params.GetString(args, "method")
	if err != nil {
		return to.ErrorResult(err)
	}
	switch method {
	case "create":
		return createBranchProtectionFn(ctx, req)
	case "edit":
		return editBranchProtectionFn(ctx, req)
	case "delete":
		return deleteBranchProtectionFn(ctx, req)
	default:
		return to.ErrorResult(fmt.Errorf("unknown method: %s", method))
	}
}

func createBranchProtectionFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	log.Debugf("Called createBranchProtectionFn")
	args := req.GetArguments()
	owner, err := params.GetString(args, "owner")
	if err != nil {
		return to.ErrorResult(err)
	}
	repo, err := params.GetString(args, "repo")
	if err != nil {
		return to.ErrorResult(err)
	}
	branchName, err := params.GetString(args, "branch_name")
	if err != nil {
		return to.ErrorResult(err)
	}
	client, err := gitea.ClientFromContext(ctx)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("get gitea client err: %v", err))
	}

	opt := gitea_sdk.CreateBranchProtectionOption{
		BranchName: branchName,
	}
	if v, ok := args["enable_push"].(bool); ok {
		opt.EnablePush = v
	}
	opt.PushWhitelistUsernames = params.GetStringSlice(args, "push_whitelist_usernames")
	opt.PushWhitelistTeams = params.GetStringSlice(args, "push_whitelist_teams")
	if v, ok := args["enable_merge_whitelist"].(bool); ok {
		opt.EnableMergeWhitelist = v
	}
	opt.MergeWhitelistUsernames = params.GetStringSlice(args, "merge_whitelist_usernames")
	opt.MergeWhitelistTeams = params.GetStringSlice(args, "merge_whitelist_teams")
	if val, exists := args["required_approvals"]; exists {
		if i, ok := params.ToInt64(val); ok {
			opt.RequiredApprovals = i
		}
	}
	if v, ok := args["block_on_rejected_reviews"].(bool); ok {
		opt.BlockOnRejectedReviews = v
	}
	if v, ok := args["block_on_outdated_branch"].(bool); ok {
		opt.BlockOnOutdatedBranch = v
	}
	if v, ok := args["dismiss_stale_approvals"].(bool); ok {
		opt.DismissStaleApprovals = v
	}
	if v, ok := args["require_signed_commits"].(bool); ok {
		opt.RequireSignedCommits = v
	}
	if v, ok := args["protected_file_patterns"].(string); ok {
		opt.ProtectedFilePatterns = v
	}
	if v, ok := args["unprotected_file_patterns"].(string); ok {
		opt.UnprotectedFilePatterns = v
	}

	bp, _, err := client.CreateBranchProtection(owner, repo, opt)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("create branch protection in %v/%v err: %v", owner, repo, err))
	}
	return to.TextResult(slimBranchProtection(bp))
}

func editBranchProtectionFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	log.Debugf("Called editBranchProtectionFn")
	args := req.GetArguments()
	owner, err := params.GetString(args, "owner")
	if err != nil {
		return to.ErrorResult(err)
	}
	repo, err := params.GetString(args, "repo")
	if err != nil {
		return to.ErrorResult(err)
	}
	ruleName, err := params.GetString(args, "rule_name")
	if err != nil {
		return to.ErrorResult(err)
	}
	client, err := gitea.ClientFromContext(ctx)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("get gitea client err: %v", err))
	}

	opt := gitea_sdk.EditBranchProtectionOption{}
	if v, ok := args["enable_push"].(bool); ok {
		opt.EnablePush = &v
	}
	if v := params.GetStringSlice(args, "push_whitelist_usernames"); v != nil {
		opt.PushWhitelistUsernames = v
	}
	if v := params.GetStringSlice(args, "push_whitelist_teams"); v != nil {
		opt.PushWhitelistTeams = v
	}
	if v, ok := args["enable_merge_whitelist"].(bool); ok {
		opt.EnableMergeWhitelist = &v
	}
	if v := params.GetStringSlice(args, "merge_whitelist_usernames"); v != nil {
		opt.MergeWhitelistUsernames = v
	}
	if v := params.GetStringSlice(args, "merge_whitelist_teams"); v != nil {
		opt.MergeWhitelistTeams = v
	}
	if val, exists := args["required_approvals"]; exists {
		if i, ok := params.ToInt64(val); ok {
			opt.RequiredApprovals = &i
		}
	}
	if v, ok := args["block_on_rejected_reviews"].(bool); ok {
		opt.BlockOnRejectedReviews = &v
	}
	if v, ok := args["block_on_outdated_branch"].(bool); ok {
		opt.BlockOnOutdatedBranch = &v
	}
	if v, ok := args["dismiss_stale_approvals"].(bool); ok {
		opt.DismissStaleApprovals = &v
	}
	if v, ok := args["require_signed_commits"].(bool); ok {
		opt.RequireSignedCommits = &v
	}
	if v, ok := args["protected_file_patterns"].(string); ok {
		opt.ProtectedFilePatterns = &v
	}
	if v, ok := args["unprotected_file_patterns"].(string); ok {
		opt.UnprotectedFilePatterns = &v
	}

	bp, _, err := client.EditBranchProtection(owner, repo, ruleName, opt)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("edit branch protection %v in %v/%v err: %v", ruleName, owner, repo, err))
	}
	return to.TextResult(slimBranchProtection(bp))
}

func deleteBranchProtectionFn(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) {
	log.Debugf("Called deleteBranchProtectionFn")
	args := req.GetArguments()
	owner, err := params.GetString(args, "owner")
	if err != nil {
		return to.ErrorResult(err)
	}
	repo, err := params.GetString(args, "repo")
	if err != nil {
		return to.ErrorResult(err)
	}
	ruleName, err := params.GetString(args, "rule_name")
	if err != nil {
		return to.ErrorResult(err)
	}
	client, err := gitea.ClientFromContext(ctx)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("get gitea client err: %v", err))
	}
	_, err = client.DeleteBranchProtection(owner, repo, ruleName)
	if err != nil {
		return to.ErrorResult(fmt.Errorf("delete branch protection %v from %v/%v err: %v", ruleName, owner, repo, err))
	}
	return mcp.NewToolResultText(fmt.Sprintf("Branch protection rule '%s' deleted from %s/%s", ruleName, owner, repo)), nil
}