From 1a887fc036d6e133797757bbf8936a4d4cecccc1 Mon Sep 17 00:00:00 2001 From: nfebe Date: Tue, 30 Jun 2026 01:30:52 +0100 Subject: [PATCH 1/4] feat(agent): Ship Albacore lifecycle fixes and the access-groups app Setting a service image or updating compose now validates a relative env_file against the deployment directory, so deployments using ./.env no longer fail validation. Additional domains route to the deployment's own container instead of an arbitrary container that happens to share a service name, fixing intermittent cross-deployment responses. Long hostnames no longer break nginx after an upgrade now that the hash sizing is applied to managed configs. Deployment and service actions can force-recreate containers, rebuild without cache, and pull fresh images, so updated environment variables and images actually take effect without dropping to the terminal. A deployment's primary service can be pinned and is reported per service, and the pin survives compose updates and re-discovery. Adds Access Groups, a built-in app that models per-deployment east-west and egress rules. Rules are persisted and previewable, with enforcement still to come. --- .../compose_validation_integration_test.go | 46 ++++- internal/api/jobs.go | 8 +- internal/api/jobs_test.go | 49 +++++- internal/api/metadata_update_test.go | 46 +++++ internal/api/server.go | 126 ++++++++++++-- internal/docker/compose.go | 97 ++++++++++- internal/docker/compose_test.go | 25 +++ internal/docker/compose_yaml.go | 49 ++++++ internal/docker/compose_yaml_test.go | 33 ++++ internal/docker/discovery.go | 12 +- internal/infra/manager.go | 39 +++++ internal/infra/manager_test.go | 45 +++++ internal/nginx/manager.go | 29 +++- internal/nginx/manager_test.go | 50 +++++- pkg/models/deployment.go | 47 ++++- pkg/models/deployment_test.go | 34 ++++ pkg/plugins/accessgroups/plugin.go | 161 ++++++++++++++++++ pkg/plugins/accessgroups/plugin_test.go | 99 +++++++++++ templates/infra/nginx/nginx.conf | 1 + templates/infra/nginx/nginx.lua.conf | 2 + 20 files changed, 956 insertions(+), 42 deletions(-) create mode 100644 pkg/plugins/accessgroups/plugin.go create mode 100644 pkg/plugins/accessgroups/plugin_test.go diff --git a/internal/api/compose_validation_integration_test.go b/internal/api/compose_validation_integration_test.go index 255ab38..6a41122 100644 --- a/internal/api/compose_validation_integration_test.go +++ b/internal/api/compose_validation_integration_test.go @@ -1,9 +1,12 @@ package api import ( + "os" + "path/filepath" "strings" "testing" + "github.com/flatrun/agent/internal/docker" "github.com/flatrun/agent/pkg/config" ) @@ -13,7 +16,7 @@ services: app: image: nginx:alpine ` - err := validateComposeWithComposeGo(validCompose) + err := validateComposeWithComposeGo(validCompose, ".") if err != nil { t.Errorf("validateComposeWithComposeGo(valid compose) = %v, want nil", err) } @@ -25,7 +28,7 @@ services: app: image: 12345 ` - err := validateComposeWithComposeGo(invalidCompose) + err := validateComposeWithComposeGo(invalidCompose, ".") if err == nil { t.Error("validateComposeWithComposeGo(invalid compose) = nil, want error") } @@ -40,7 +43,7 @@ services: app: image: [broken yaml ` - err := validateComposeWithComposeGo(invalidCompose) + err := validateComposeWithComposeGo(invalidCompose, ".") if err == nil { t.Error("validateComposeWithComposeGo(invalid YAML) = nil, want error") } @@ -69,3 +72,40 @@ networks: t.Errorf("validateComposeContent(valid) = %v, want nil", err) } } + +// A compose with a relative env_file must validate against the deployment directory, +// where the file lives, not the agent's working directory. +func TestValidateComposeContent_RelativeEnvFile_ResolvesAgainstDeploymentDir(t *testing.T) { + base := t.TempDir() + const name = "envfileapp" + deployDir := filepath.Join(base, name) + if err := os.MkdirAll(deployDir, 0755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(deployDir, ".env"), []byte("FOO=bar\n"), 0644); err != nil { + t.Fatal(err) + } + + s := &Server{ + config: &config.Config{ + Infrastructure: config.InfrastructureConfig{DefaultProxyNetwork: "proxy"}, + }, + manager: docker.NewManager(base), + } + + compose := `name: envfileapp +services: + app: + image: nginx:alpine + env_file: + - ./.env + networks: + - proxy +networks: + proxy: + external: true +` + if err := s.validateComposeContent(compose, name); err != nil { + t.Errorf("validateComposeContent with relative env_file in deployment dir = %v, want nil", err) + } +} diff --git a/internal/api/jobs.go b/internal/api/jobs.go index eca15e7..6347c0f 100644 --- a/internal/api/jobs.go +++ b/internal/api/jobs.go @@ -30,6 +30,7 @@ type ActionJob struct { deployment string service string action string + opts actionOptions activeKey string mu sync.Mutex @@ -163,14 +164,14 @@ func newJobRegistry() *jobRegistry { // create registers a new pending job for the deployment. If an action is // already in flight for that deployment it returns the existing job and false, // so the caller can reject the request and point the client at the live job. -func (r *jobRegistry) create(deployment, action string) (*ActionJob, bool) { - return r.createScoped(deployment, "", action) +func (r *jobRegistry) create(deployment, action string, opts actionOptions) (*ActionJob, bool) { + return r.createScoped(deployment, "", action, opts) } // createScoped registers a job serialized on the deployment, or on a single // service within it when service is set, so two different services can act // concurrently while the same service cannot. -func (r *jobRegistry) createScoped(deployment, service, action string) (*ActionJob, bool) { +func (r *jobRegistry) createScoped(deployment, service, action string, opts actionOptions) (*ActionJob, bool) { r.mu.Lock() defer r.mu.Unlock() @@ -193,6 +194,7 @@ func (r *jobRegistry) createScoped(deployment, service, action string) (*ActionJ deployment: deployment, service: service, action: action, + opts: opts, activeKey: key, status: JobPending, startedAt: time.Now(), diff --git a/internal/api/jobs_test.go b/internal/api/jobs_test.go index edffca8..1648073 100644 --- a/internal/api/jobs_test.go +++ b/internal/api/jobs_test.go @@ -16,12 +16,16 @@ import ( // fakeRunner stands in for the docker-backed action runner so tests exercise // the job lifecycle over HTTP without shelling out to compose. type fakeRunner struct { - lines []string - err error - gate chan struct{} + lines []string + err error + gate chan struct{} + optsCh chan actionOptions } -func (f *fakeRunner) run(_, _ string, emit func(string)) error { +func (f *fakeRunner) run(_, _ string, opts actionOptions, emit func(string)) error { + if f.optsCh != nil { + f.optsCh <- opts + } for _, l := range f.lines { emit(l) } @@ -35,8 +39,8 @@ func newJobTestServer(runner *fakeRunner) *Server { gin.SetMode(gin.TestMode) s := &Server{jobs: newJobRegistry()} s.runDeploymentAction = runner.run - s.runServiceAction = func(action, name, service string, emit func(string)) error { - return runner.run(action, name, emit) + s.runServiceAction = func(action, name, service string, opts actionOptions, emit func(string)) error { + return runner.run(action, name, opts, emit) } return s } @@ -215,6 +219,37 @@ func TestServiceRestartJobIsScopedPerService(t *testing.T) { } } +// Effective-apply flags in the request body must reach the runner so updated env vars +// and images take effect. +func TestServiceJobThreadsEffectiveApplyOptions(t *testing.T) { + optsCh := make(chan actionOptions, 1) + s := newJobTestServer(&fakeRunner{lines: []string{"ok"}, optsCh: optsCh}) + srv := newSkippableHTTPServer(t, newJobRouter(s)) + defer srv.Close() + + resp, err := http.Post( + srv.URL+"/api/deployments/app/services/web/job", + "application/json", + strings.NewReader(`{"action":"rebuild","force_recreate":true,"no_cache":true,"fresh_pull":true}`), + ) + if err != nil { + t.Fatalf("service job request failed: %v", err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusAccepted { + t.Fatalf("expected 202, got %d", resp.StatusCode) + } + + select { + case got := <-optsCh: + if !got.ForceRecreate || !got.NoCache || !got.FreshPull { + t.Fatalf("runner received opts %+v, want all flags set", got) + } + case <-time.After(5 * time.Second): + t.Fatal("runner was not invoked with options") + } +} + func TestDeploymentJobStreamReplaysAndCompletes(t *testing.T) { s := &Server{ jobs: newJobRegistry(), @@ -222,7 +257,7 @@ func TestDeploymentJobStreamReplaysAndCompletes(t *testing.T) { } // Pre-populate a finished job so the stream must replay buffered output. - job, _ := s.jobs.create("myapp", "start") + job, _ := s.jobs.create("myapp", "start", actionOptions{}) job.appendLine("Pulling nginx") job.appendLine("Started") s.jobs.finish(job, JobSucceeded, "") diff --git a/internal/api/metadata_update_test.go b/internal/api/metadata_update_test.go index 43b652d..c430690 100644 --- a/internal/api/metadata_update_test.go +++ b/internal/api/metadata_update_test.go @@ -78,6 +78,52 @@ func TestMergeMetadata_PartialUpdatePreservesOtherFields(t *testing.T) { } } +// Pinning a primary service records the pin and syncs the default-domain upstream even +// when the networking block is not part of the same update. +func TestMergeMetadata_PrimaryServiceSyncsRoutingService(t *testing.T) { + existing := &models.ServiceMetadata{ + Name: "shop", + Networking: models.NetworkingConfig{ + Expose: true, + Domain: "shop.example.com", + Service: "web", + ContainerPort: 80, + }, + } + + sentFields, incoming := parseTestJSON(t, `{"primary_service": "api"}`) + merged := mergeMetadata(existing, &incoming, sentFields) + + if merged.PrimaryService != "api" { + t.Errorf("PrimaryService = %q, want %q", merged.PrimaryService, "api") + } + if merged.Networking.Service != "api" { + t.Errorf("default-domain Service should follow the pin: got %q, want %q", merged.Networking.Service, "api") + } + if merged.Networking.Domain != "shop.example.com" { + t.Error("unsent networking fields should be preserved") + } +} + +// An access_groups update is persisted through the metadata merge, while unsent fields +// are preserved. +func TestMergeMetadata_AccessGroups(t *testing.T) { + existing := &models.ServiceMetadata{Name: "shop", Type: "laravel"} + + sentFields, incoming := parseTestJSON(t, `{"access_groups": {"enabled": true, "egress": "deny-all", "allow": [{"to": "db", "port": 5432}]}}`) + merged := mergeMetadata(existing, &incoming, sentFields) + + if merged.AccessGroups == nil || !merged.AccessGroups.Enabled { + t.Fatalf("access_groups should be set and enabled, got %+v", merged.AccessGroups) + } + if merged.AccessGroups.Egress != "deny-all" || len(merged.AccessGroups.Allow) != 1 || merged.AccessGroups.Allow[0].To != "db" { + t.Errorf("access_groups not merged correctly: %+v", merged.AccessGroups) + } + if merged.Type != "laravel" { + t.Error("unsent field Type should be preserved") + } +} + func TestMergeMetadata_SentFieldOverwritesExisting(t *testing.T) { existing := &models.ServiceMetadata{ Name: "old-name", diff --git a/internal/api/server.go b/internal/api/server.go index 1c4d161..366847d 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -47,6 +47,7 @@ import ( "github.com/flatrun/agent/pkg/config" "github.com/flatrun/agent/pkg/models" "github.com/flatrun/agent/pkg/plugins" + "github.com/flatrun/agent/pkg/plugins/accessgroups" dnsPlugins "github.com/flatrun/agent/pkg/plugins/dns" "github.com/flatrun/agent/pkg/subdomain" "github.com/flatrun/agent/pkg/version" @@ -66,6 +67,7 @@ type Server struct { certsDiscovery *certs.Discovery networksManager *networks.Manager pluginRegistry *plugins.Registry + accessGroups *accessgroups.Plugin authMiddleware *auth.Middleware authManager *auth.Manager proxyOrchestrator *proxy.Orchestrator @@ -93,9 +95,9 @@ type Server struct { jobs *jobRegistry // runDeploymentAction runs a deployment action and streams each output // line to emit. Overridable in tests so they need not shell out to docker. - runDeploymentAction func(action, name string, emit func(line string)) error + runDeploymentAction func(action, name string, opts actionOptions, emit func(line string)) error // runServiceAction runs an action on a single service, streaming output. - runServiceAction func(action, name, service string, emit func(line string)) error + runServiceAction func(action, name, service string, opts actionOptions, emit func(line string)) error statsMu sync.RWMutex statsCache gin.H @@ -155,6 +157,20 @@ func New(cfg *config.Config, configPath string) *Server { pluginsDir := filepath.Join(cfg.DeploymentsPath, ".flatrun", "plugins") pluginRegistry := plugins.NewRegistry(pluginsDir) _ = pluginRegistry.LoadFromDisk() + + // Access Groups is a built-in app: register it so it lists as an installed plugin and + // can serve its routes. Its plan endpoint reads the deployment's stored policy. + accessGroupsPlugin := accessgroups.New(func(name string) (*models.AccessGroupsConfig, error) { + dep, err := manager.GetDeployment(name) + if err != nil { + return nil, err + } + if dep.Metadata == nil { + return nil, nil + } + return dep.Metadata.AccessGroups, nil + }) + _ = pluginRegistry.Register(accessGroupsPlugin) authMiddleware := auth.NewMiddleware(&cfg.Auth) setupManager := setup.NewManager(cfg, configPath) @@ -269,6 +285,7 @@ func New(cfg *config.Config, configPath string) *Server { certsDiscovery: certsDiscovery, networksManager: networksManager, pluginRegistry: pluginRegistry, + accessGroups: accessGroupsPlugin, authMiddleware: authMiddleware, authManager: authManager, proxyOrchestrator: proxyOrchestrator, @@ -686,6 +703,9 @@ func (s *Server) setupRoutes() { NewPowerDNSHandlers(s.powerDNSManager).RegisterRoutes(protected) } + // Access Groups built-in app routes (read-only plan; enforcement not wired yet) + _ = s.accessGroups.RegisterRoutes(protected) + // Cluster endpoints clusterGroup := protected.Group("/cluster") clusterGroup.Use(s.authMiddleware.RequirePermission(auth.PermClusterRead)) @@ -722,6 +742,15 @@ func (s *Server) Start() error { Handler: s.router, } + // Refresh a managed base nginx config so an upgraded agent picks up template changes + // (e.g. server_names_hash sizing) without waiting on a security toggle. Best-effort and + // off the startup path since it may reload nginx once the container is ready. + go func() { + if err := s.infraManager.EnsureBaseNginxConfig(); err != nil { + log.Printf("[infra] failed to refresh base nginx config on startup: %v", err) + } + }() + if s.config.Certbot.Enabled && s.config.Certbot.AutoRenewalEnabled { s.certRenewer = ssl.NewRenewer( s.proxyOrchestrator.SSLManager(), @@ -1692,6 +1721,14 @@ func mergeMetadata(existing, incoming *models.ServiceMetadata, sentFields map[st if _, ok := sentFields["credential_id"]; ok { merged.CredentialID = incoming.CredentialID } + if _, ok := sentFields["primary_service"]; ok { + merged.PrimaryService = incoming.PrimaryService + // Keep the default-domain upstream in sync so routing follows the pin even + // when the networking block is not part of this update. + if incoming.PrimaryService != "" { + merged.Networking.Service = incoming.PrimaryService + } + } if _, ok := sentFields["networking"]; ok { merged.Networking = incoming.Networking } @@ -1707,6 +1744,9 @@ func mergeMetadata(existing, incoming *models.ServiceMetadata, sentFields map[st if _, ok := sentFields["security"]; ok { merged.Security = incoming.Security } + if _, ok := sentFields["access_groups"]; ok { + merged.AccessGroups = incoming.AccessGroups + } if _, ok := sentFields["backup"]; ok { merged.Backup = incoming.Backup } @@ -1824,7 +1864,11 @@ func (s *Server) rebuildDeployment(c *gin.Context) { func (s *Server) enqueueDeploymentAction(c *gin.Context, action string) { name := c.Param("name") - job, created := s.jobs.create(name, action) + // The options body is optional; an empty or non-JSON body simply means no flags. + var opts actionOptions + _ = c.ShouldBindJSON(&opts) + + job, created := s.jobs.create(name, action, opts) if !created { c.JSON(http.StatusConflict, gin.H{ "error": "An action is already running for this deployment", @@ -1845,7 +1889,7 @@ func (s *Server) enqueueDeploymentAction(c *gin.Context, action string) { func (s *Server) runActionJob(job *ActionJob) { job.setRunning() - err := s.runDeploymentAction(job.action, job.deployment, job.appendLine) + err := s.runDeploymentAction(job.action, job.deployment, job.opts, job.appendLine) if err != nil { s.jobs.finish(job, JobFailed, err.Error()) return @@ -1853,11 +1897,35 @@ func (s *Server) runActionJob(job *ActionJob) { s.jobs.finish(job, JobSucceeded, "") } -func (s *Server) defaultRunDeploymentAction(action, name string, emit func(line string)) error { +// actionOptions carries the optional effective-apply flags for a deployment or service +// action, so updated environment variables and images actually take effect instead of a +// plain start/restart reusing cached config and images. +type actionOptions struct { + ForceRecreate bool `json:"force_recreate"` + NoCache bool `json:"no_cache"` + FreshPull bool `json:"fresh_pull"` +} + +func (o actionOptions) runOptions() []docker.RunOption { + var opts []docker.RunOption + if o.ForceRecreate { + opts = append(opts, docker.WithForceRecreate()) + } + if o.NoCache { + opts = append(opts, docker.WithNoCache()) + } + if o.FreshPull { + opts = append(opts, docker.WithFreshPull()) + } + return opts +} + +func (s *Server) defaultRunDeploymentAction(action, name string, actOpts actionOptions, emit func(line string)) error { authCfg, opts := s.deploymentAuthOptions(name) defer authCfg.Close() opts = append(opts, docker.WithLineSink(emit)) + opts = append(opts, actOpts.runOptions()...) var err error switch action { @@ -1884,7 +1952,10 @@ var streamableServiceActions = map[string]bool{ // taken from the body so one route covers every action. func (s *Server) enqueueServiceJob(c *gin.Context) { var req struct { - Action string `json:"action"` + Action string `json:"action"` + ForceRecreate bool `json:"force_recreate"` + NoCache bool `json:"no_cache"` + FreshPull bool `json:"fresh_pull"` } if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid request body: " + err.Error()}) @@ -1898,7 +1969,8 @@ func (s *Server) enqueueServiceJob(c *gin.Context) { name := c.Param("name") service := c.Param("service") - job, created := s.jobs.createScoped(name, service, req.Action) + opts := actionOptions{ForceRecreate: req.ForceRecreate, NoCache: req.NoCache, FreshPull: req.FreshPull} + job, created := s.jobs.createScoped(name, service, req.Action, opts) if !created { c.JSON(http.StatusConflict, gin.H{ "error": "An action is already running for this service", @@ -1920,7 +1992,7 @@ func (s *Server) enqueueServiceJob(c *gin.Context) { func (s *Server) runServiceActionJob(job *ActionJob) { job.setRunning() - err := s.runServiceAction(job.action, job.deployment, job.service, job.appendLine) + err := s.runServiceAction(job.action, job.deployment, job.service, job.opts, job.appendLine) if err != nil { s.jobs.finish(job, JobFailed, err.Error()) return @@ -1928,11 +2000,12 @@ func (s *Server) runServiceActionJob(job *ActionJob) { s.jobs.finish(job, JobSucceeded, "") } -func (s *Server) defaultRunServiceAction(action, name, service string, emit func(line string)) error { +func (s *Server) defaultRunServiceAction(action, name, service string, actOpts actionOptions, emit func(line string)) error { authCfg, opts := s.deploymentAuthOptions(name) defer authCfg.Close() opts = append(opts, docker.WithLineSink(emit)) + opts = append(opts, actOpts.runOptions()...) var err error switch action { @@ -3763,7 +3836,7 @@ func (s *Server) inferRegistryHostFromCompose(content string) string { return "" } -func (s *Server) validateComposeContent(content, _ string) error { +func (s *Server) validateComposeContent(content, name string) error { var compose composeFile if err := yaml.Unmarshal([]byte(content), &compose); err != nil { return fmt.Errorf("invalid YAML syntax: %w", err) @@ -3801,24 +3874,42 @@ func (s *Server) validateComposeContent(content, _ string) error { } } - if err := validateComposeWithComposeGo(content); err != nil { + if err := validateComposeWithComposeGo(content, s.composeValidationDir(name)); err != nil { return err } return nil } -func validateComposeWithComposeGo(content string) error { +// composeValidationDir returns the directory that relative compose paths (such as +// a relative env_file) should resolve against during validation. For an existing +// deployment that is the deployment directory; when it does not yet exist (the +// create path) it falls back to the agent's working directory so create keeps working. +func (s *Server) composeValidationDir(name string) string { + if name == "" || s.manager == nil { + return "." + } + dir := filepath.Join(s.manager.BasePath(), name) + if info, err := os.Stat(dir); err != nil || !info.IsDir() { + return "." + } + return dir +} + +func validateComposeWithComposeGo(content, workingDir string) error { configDetails := composetypes.ConfigDetails{ ConfigFiles: []composetypes.ConfigFile{{ Filename: "compose.yml", Content: []byte(content), }}, - WorkingDir: ".", + WorkingDir: workingDir, Environment: map[string]string{}, } _, err := loader.LoadWithContext(context.Background(), configDetails, func(o *loader.Options) { - o.ResolvePaths = false + // Resolve relative paths (notably a relative env_file) against WorkingDir so an + // existing deployment's ./.env is found in the deployment directory rather than + // being read relative to the agent's own working directory. + o.ResolvePaths = true o.SkipConsistencyCheck = true }) if err != nil { @@ -4844,6 +4935,13 @@ func (s *Server) listDeploymentServices(c *gin.Context) { return } + if deployment, err := s.manager.GetDeployment(name); err == nil && deployment.Metadata != nil { + primary := deployment.Metadata.EffectivePrimaryService() + for i := range services { + services[i].IsPrimary = primary != "" && services[i].Name == primary + } + } + c.JSON(http.StatusOK, gin.H{"services": services}) } diff --git a/internal/docker/compose.go b/internal/docker/compose.go index f27ba26..20da769 100644 --- a/internal/docker/compose.go +++ b/internal/docker/compose.go @@ -25,8 +25,48 @@ func NewComposeExecutor(basePath string) *ComposeExecutor { type RunOption func(*runOpts) type runOpts struct { - extraEnv []string - lineSink func(string) + extraEnv []string + lineSink func(string) + forceRecreate bool + noCache bool + freshPull bool +} + +// WithForceRecreate recreates containers even when their config and image are +// unchanged, so updated environment variables take effect. +func WithForceRecreate() RunOption { + return func(o *runOpts) { o.forceRecreate = true } +} + +// WithNoCache rebuilds images without using the build cache before bringing the +// deployment up. +func WithNoCache() RunOption { + return func(o *runOpts) { o.noCache = true } +} + +// WithFreshPull forces images to be pulled rather than served from the local +// cache during an up. +func WithFreshPull() RunOption { + return func(o *runOpts) { o.freshPull = true } +} + +func resolveRunOpts(opts []RunOption) runOpts { + var ro runOpts + for _, opt := range opts { + opt(&ro) + } + return ro +} + +// applyUpFlags appends the effective-apply flags supported by `compose up`. +func applyUpFlags(args []string, ro runOpts) []string { + if ro.forceRecreate { + args = append(args, "--force-recreate") + } + if ro.freshPull { + args = append(args, "--pull", "always") + } + return args } func WithDockerConfig(dir string) RunOption { @@ -46,7 +86,14 @@ func WithLineSink(sink func(string)) RunOption { } func (c *ComposeExecutor) Up(deploymentPath string, opts ...RunOption) (string, error) { - return c.runCompose(deploymentPath, opts, "up", "-d", "--remove-orphans") + ro := resolveRunOpts(opts) + if ro.noCache { + if out, err := c.runCompose(deploymentPath, opts, "build", "--no-cache"); err != nil { + return out, err + } + } + args := applyUpFlags([]string{"up", "-d", "--remove-orphans"}, ro) + return c.runCompose(deploymentPath, opts, args...) } func (c *ComposeExecutor) Down(deploymentPath string, opts ...RunOption) (string, error) { @@ -66,16 +113,43 @@ func (c *ComposeExecutor) Stop(deploymentPath string, opts ...RunOption) (string } func (c *ComposeExecutor) Restart(deploymentPath string, opts ...RunOption) (string, error) { + ro := resolveRunOpts(opts) _, _ = c.runCompose(deploymentPath, opts, "down", "--remove-orphans") - return c.runCompose(deploymentPath, opts, "up", "-d", "--remove-orphans") + if ro.noCache { + if out, err := c.runCompose(deploymentPath, opts, "build", "--no-cache"); err != nil { + return out, err + } + } + args := applyUpFlags([]string{"up", "-d", "--remove-orphans"}, ro) + return c.runCompose(deploymentPath, opts, args...) } func (c *ComposeExecutor) Rebuild(deploymentPath string, opts ...RunOption) (string, error) { + ro := resolveRunOpts(opts) _, _ = c.runCompose(deploymentPath, opts, "down", "--remove-orphans") - return c.runCompose(deploymentPath, opts, "up", "-d", "--build", "--remove-orphans") + if ro.noCache { + if out, err := c.runCompose(deploymentPath, opts, "build", "--no-cache"); err != nil { + return out, err + } + } + args := applyUpFlags([]string{"up", "-d", "--build", "--remove-orphans"}, ro) + return c.runCompose(deploymentPath, opts, args...) } func (c *ComposeExecutor) StartService(deploymentPath, service string, opts ...RunOption) (string, error) { + ro := resolveRunOpts(opts) + // When an effective-apply option is set, go straight to `up` so the flags take effect; + // a plain `start` cannot recreate, rebuild, or pull. + if ro.forceRecreate || ro.freshPull || ro.noCache { + if ro.noCache { + if out, err := c.runCompose(deploymentPath, opts, "build", "--no-cache", service); err != nil { + return out, err + } + } + args := applyUpFlags([]string{"up", "-d", "--no-deps"}, ro) + args = append(args, service) + return c.runCompose(deploymentPath, opts, args...) + } output, err := c.runCompose(deploymentPath, opts, "start", service) if err != nil { return c.runCompose(deploymentPath, opts, "up", "-d", "--no-deps", service) @@ -92,7 +166,18 @@ func (c *ComposeExecutor) RestartService(deploymentPath, service string, opts .. } func (c *ComposeExecutor) RebuildService(deploymentPath, service string, opts ...RunOption) (string, error) { - return c.runCompose(deploymentPath, opts, "up", "-d", "--no-deps", "--build", "--force-recreate", service) + ro := resolveRunOpts(opts) + if ro.noCache { + if out, err := c.runCompose(deploymentPath, opts, "build", "--no-cache", service); err != nil { + return out, err + } + } + args := []string{"up", "-d", "--no-deps", "--build", "--force-recreate"} + if ro.freshPull { + args = append(args, "--pull", "always") + } + args = append(args, service) + return c.runCompose(deploymentPath, opts, args...) } func (c *ComposeExecutor) PullService(deploymentPath, service string, opts ...RunOption) (string, error) { diff --git a/internal/docker/compose_test.go b/internal/docker/compose_test.go index ffa7f18..eddce88 100644 --- a/internal/docker/compose_test.go +++ b/internal/docker/compose_test.go @@ -3,9 +3,34 @@ package docker import ( "os" "path/filepath" + "strings" "testing" ) +func TestApplyUpFlags(t *testing.T) { + base := []string{"up", "-d", "--remove-orphans"} + + if got := applyUpFlags(base, runOpts{}); strings.Join(got, " ") != "up -d --remove-orphans" { + t.Errorf("no options should not add flags, got %v", got) + } + + got := applyUpFlags(base, runOpts{forceRecreate: true, freshPull: true}) + joined := strings.Join(got, " ") + if !strings.Contains(joined, "--force-recreate") { + t.Errorf("forceRecreate should add --force-recreate, got %v", got) + } + if !strings.Contains(joined, "--pull always") { + t.Errorf("freshPull should add --pull always, got %v", got) + } +} + +func TestResolveRunOpts(t *testing.T) { + ro := resolveRunOpts([]RunOption{WithForceRecreate(), WithNoCache(), WithFreshPull()}) + if !ro.forceRecreate || !ro.noCache || !ro.freshPull { + t.Errorf("resolveRunOpts did not set all flags: %+v", ro) + } +} + func TestComposeExecutorPull(t *testing.T) { tmpDir, err := os.MkdirTemp("", "compose-pull-test-*") if err != nil { diff --git a/internal/docker/compose_yaml.go b/internal/docker/compose_yaml.go index af1c211..03d8e4d 100644 --- a/internal/docker/compose_yaml.go +++ b/internal/docker/compose_yaml.go @@ -157,6 +157,55 @@ func ParseComposeYAML(content string) (map[string]interface{}, error) { return compose, nil } +// ContainerNameForService returns the DNS-resolvable container name for a service in +// a deployment, matching what EnsureContainerNames assigns. An explicit container_name in +// the compose wins; otherwise the EnsureContainerNames rule applies (primary service -> +// deploymentName, all others -> "{deploymentName}-{service}"). When the compose cannot be +// parsed it falls back to the deployment-scoped name so the upstream is never the bare, +// collision-prone service name shared across deployments. +func ContainerNameForService(content, deploymentName, service string) string { + if deploymentName == "" || service == "" { + return service + } + scoped := fmt.Sprintf("%s-%s", deploymentName, service) + + compose, err := ParseComposeYAML(content) + if err != nil { + if service == "app" { + return deploymentName + } + return scoped + } + services, ok := compose["services"].(map[string]interface{}) + if !ok || len(services) == 0 { + if service == "app" { + return deploymentName + } + return scoped + } + + if svc, ok := services[service].(map[string]interface{}); ok { + if cn, ok := svc["container_name"].(string); ok && cn != "" { + return cn + } + } + + var primaryService string + for name := range services { + if name == "app" { + primaryService = "app" + break + } + if primaryService == "" { + primaryService = name + } + } + if service == primaryService { + return deploymentName + } + return scoped +} + // EnsureContainerNames ensures all services have explicit container_name set. // The primary service (preferring "app") gets deploymentName as its container name. // All other services get "{deploymentName}-{serviceName}". diff --git a/internal/docker/compose_yaml_test.go b/internal/docker/compose_yaml_test.go index 194497e..f0f98eb 100644 --- a/internal/docker/compose_yaml_test.go +++ b/internal/docker/compose_yaml_test.go @@ -625,3 +625,36 @@ func TestEnsureContainerNames_NoServices(t *testing.T) { t.Errorf("no services should return unchanged content") } } + +func TestContainerNameForService(t *testing.T) { + const multi = `name: shop +services: + app: + image: nginx + container_name: shop + worker: + image: redis + container_name: shop-worker +` + tests := []struct { + name string + content string + deployment string + service string + want string + }{ + {"explicit container_name for primary", multi, "shop", "app", "shop"}, + {"explicit container_name for non-primary", multi, "shop", "worker", "shop-worker"}, + {"missing compose falls back to scoped name", "", "shop", "worker", "shop-worker"}, + {"missing compose treats app as primary", "", "shop", "app", "shop"}, + {"empty deployment returns bare service", multi, "", "app", "app"}, + {"no explicit name uses primary rule", "name: x\nservices:\n app:\n image: nginx\n api:\n image: go\n", "x", "api", "x-api"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := ContainerNameForService(tt.content, tt.deployment, tt.service); got != tt.want { + t.Errorf("ContainerNameForService(%q, %q) = %q, want %q", tt.deployment, tt.service, got, tt.want) + } + }) + } +} diff --git a/internal/docker/discovery.go b/internal/docker/discovery.go index c20b657..998d220 100644 --- a/internal/docker/discovery.go +++ b/internal/docker/discovery.go @@ -140,6 +140,12 @@ func (d *Discovery) GetDeployment(name string) (*models.Deployment, error) { } func (d *Discovery) findComposeFile(dirPath string) string { + return FindComposeFile(dirPath) +} + +// FindComposeFile returns the path to the compose file in dirPath, preferring the +// standard filenames and falling back to a *compose*.yml/yaml glob. Returns "" when none. +func FindComposeFile(dirPath string) string { // First check exact standard names (preferred) standardNames := []string{ "docker-compose.yml", @@ -579,7 +585,11 @@ func (d *Discovery) UpdateComposeFile(name string, content string) error { existing, err := d.loadMetadata(metadataPath) if err == nil { existing.Networking.ContainerPort = newMeta.Networking.ContainerPort - if newMeta.Networking.Service != "" { + // Respect a user-pinned primary service; only let auto-detection set the + // routing service when the user has not explicitly chosen one. + if existing.PrimaryService != "" { + existing.Networking.Service = existing.PrimaryService + } else if newMeta.Networking.Service != "" { existing.Networking.Service = newMeta.Networking.Service } if err := d.SaveMetadata(name, existing); err != nil { diff --git a/internal/infra/manager.go b/internal/infra/manager.go index 74ab02f..a8e7b0d 100644 --- a/internal/infra/manager.go +++ b/internal/infra/manager.go @@ -586,6 +586,45 @@ func (m *Manager) getNginxDir() string { return filepath.Dir(configPath) } +// EnsureBaseNginxConfig refreshes an already-managed base nginx.conf from the current +// template so an upgraded agent picks up template changes (such as the server_names_hash +// settings) without a security toggle. It preserves the on-disk variant (lua or plain) and +// is a no-op when no managed config exists or the content already matches. +func (m *Manager) EnsureBaseNginxConfig() error { + m.mu.Lock() + defer m.mu.Unlock() + + nginxDir := m.getNginxDir() + if nginxDir == "" { + return nil + } + confPath := filepath.Join(nginxDir, "nginx.conf") + existing, err := os.ReadFile(confPath) + if err != nil { + // No managed base config on disk; do not introduce one. + return nil + } + + luaEnabled := strings.Contains(string(existing), "lua_package_path") + nginxConf, err := templates.GetNginxConfigWithData(luaEnabled, templates.NginxConfigData{ + RejectUnknownDomains: m.config.Nginx.RejectUnknownDomains, + }) + if err != nil { + return err + } + if string(nginxConf) == string(existing) { + return nil + } + + if err := os.WriteFile(confPath, nginxConf, 0644); err != nil { + return err + } + // Best-effort reload: the rewritten config also takes effect on the next nginx + // restart, so a reload failure here (e.g. container not yet ready) is not fatal. + _ = m.reloadNginx() + return nil +} + // SecurityHealthCheck represents the result of a security setup health check type SecurityHealthCheck struct { Status string `json:"status"` diff --git a/internal/infra/manager_test.go b/internal/infra/manager_test.go index 5ce4208..2eaffbc 100644 --- a/internal/infra/manager_test.go +++ b/internal/infra/manager_test.go @@ -211,3 +211,48 @@ func TestGetNginxDir(t *testing.T) { }) } } + +// An upgraded agent must refresh an existing managed nginx.conf so it gains the +// server_names_hash sizing without a security toggle, and must not create one where +// none exists. +func TestEnsureBaseNginxConfig(t *testing.T) { + t.Run("refreshes an existing managed config", func(t *testing.T) { + nginxDir := t.TempDir() + confPath := filepath.Join(nginxDir, "nginx.conf") + // An older managed (lua) config that predates the server_names_hash settings. + old := "http {\n lua_package_path \"/etc/nginx/lua/?.lua;;\";\n types_hash_max_size 2048;\n}\n" + if err := os.WriteFile(confPath, []byte(old), 0644); err != nil { + t.Fatal(err) + } + + cfg := &config.Config{ + DeploymentsPath: nginxDir, + Nginx: config.NginxConfig{ConfigPath: filepath.Join(nginxDir, "conf.d")}, + } + if err := NewManager(cfg).EnsureBaseNginxConfig(); err != nil { + t.Fatalf("EnsureBaseNginxConfig() = %v", err) + } + + content, err := os.ReadFile(confPath) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(content), "server_names_hash_bucket_size") { + t.Errorf("refreshed config should contain server_names_hash_bucket_size, got:\n%s", content) + } + }) + + t.Run("does not create a config where none exists", func(t *testing.T) { + nginxDir := t.TempDir() + cfg := &config.Config{ + DeploymentsPath: nginxDir, + Nginx: config.NginxConfig{ConfigPath: filepath.Join(nginxDir, "conf.d")}, + } + if err := NewManager(cfg).EnsureBaseNginxConfig(); err != nil { + t.Fatalf("EnsureBaseNginxConfig() = %v", err) + } + if _, err := os.Stat(filepath.Join(nginxDir, "nginx.conf")); !os.IsNotExist(err) { + t.Errorf("EnsureBaseNginxConfig must not create a managed config where none existed") + } + }) +} diff --git a/internal/nginx/manager.go b/internal/nginx/manager.go index 5686d5f..73039c5 100644 --- a/internal/nginx/manager.go +++ b/internal/nginx/manager.go @@ -13,6 +13,7 @@ import ( "text/template" "time" + "github.com/flatrun/agent/internal/docker" "github.com/flatrun/agent/pkg/config" "github.com/flatrun/agent/pkg/models" ) @@ -523,7 +524,7 @@ func (m *Manager) generateMultiDomainConfig(deployment *models.Deployment) (stri } } - servers := m.groupDomainsByHost(domains, deployment.Name) + servers := m.groupDomainsByHost(domains, deployment.Name, m.deploymentComposeContent(deployment)) data := multiRouteTemplateData{ DeploymentName: deployment.Name, @@ -566,7 +567,26 @@ func (m *Manager) generateMultiDomainConfig(deployment *models.Deployment) (stri return buf.String(), nil } -func (m *Manager) groupDomainsByHost(domains []models.DomainConfig, deploymentName string) []serverData { +// deploymentComposeContent reads the deployment's compose file so the upstream for an +// additional domain can be resolved to the service's unique container name. Returns "" when +// the compose cannot be read; ContainerNameForService then falls back to a scoped name. +func (m *Manager) deploymentComposeContent(deployment *models.Deployment) string { + dir := deployment.Path + if dir == "" { + dir = filepath.Join(m.basePath, deployment.Name) + } + path := docker.FindComposeFile(dir) + if path == "" { + return "" + } + data, err := os.ReadFile(path) + if err != nil { + return "" + } + return string(data) +} + +func (m *Manager) groupDomainsByHost(domains []models.DomainConfig, deploymentName, composeContent string) []serverData { hostDomains := make(map[string][]models.DomainConfig) for _, d := range domains { hostDomains[d.Domain] = append(hostDomains[d.Domain], d) @@ -596,10 +616,15 @@ func (m *Manager) groupDomainsByHost(domains []models.DomainConfig, deploymentNa } seenPaths[path] = true + // Route to the service's unique container name, not the bare Compose service + // name (which is not unique across deployments sharing the proxy network and + // resolves via embedded DNS to an arbitrary deployment's container). service := d.Service if service == "" { log.Printf("[proxy] warning: domain %q has no service set for deployment %q, falling back to deployment name", d.Domain, deploymentName) service = deploymentName + } else { + service = docker.ContainerNameForService(composeContent, deploymentName, service) } port := d.ContainerPort diff --git a/internal/nginx/manager_test.go b/internal/nginx/manager_test.go index 006ba0b..4dde269 100644 --- a/internal/nginx/manager_test.go +++ b/internal/nginx/manager_test.go @@ -2027,8 +2027,8 @@ func TestGenerateMultiDomainConfig_ContainerPort(t *testing.T) { t.Fatalf("generateMultiDomainConfig failed: %v", err) } - if !strings.Contains(config, "web:9090") { - t.Errorf("config should proxy to port 9090, got:\n%s", config) + if !strings.Contains(config, "port-test-web:9090") { + t.Errorf("config should proxy to the service's unique container name on port 9090, got:\n%s", config) } }) @@ -2051,7 +2051,7 @@ func TestGenerateMultiDomainConfig_ContainerPort(t *testing.T) { t.Fatalf("generateMultiDomainConfig failed: %v", err) } - if !strings.Contains(config, "web:80") { + if !strings.Contains(config, "port-default-test-web:80") { t.Errorf("config should default to port 80 when ContainerPort is 0, got:\n%s", config) } }) @@ -2198,7 +2198,7 @@ func TestGroupDomainsByHost_DeduplicatesLocations(t *testing.T) { {ID: "3", Domain: "example.com", PathPrefix: "/api", ContainerPort: 3000, Service: "backend"}, } - servers := m.groupDomainsByHost(domains, "test-app") + servers := m.groupDomainsByHost(domains, "test-app", "") if len(servers) != 1 { t.Fatalf("expected 1 server, got %d", len(servers)) @@ -2217,3 +2217,45 @@ func TestGroupDomainsByHost_DeduplicatesLocations(t *testing.T) { t.Errorf("expected exactly 1 location for '/', got %d", pathCounts["/"]) } } + +// An additional domain must proxy to the service's unique container name (read from the +// deployment's compose), never the bare service name that collides across deployments +// sharing the proxy network. +func TestGenerateMultiDomainConfig_UsesUniqueContainerName(t *testing.T) { + dir := t.TempDir() + compose := `name: tenant-a +services: + app: + image: nginx + container_name: tenant-a + api: + image: go + container_name: tenant-a-api +` + if err := os.WriteFile(filepath.Join(dir, "docker-compose.yml"), []byte(compose), 0644); err != nil { + t.Fatal(err) + } + + m := NewManager(&config.NginxConfig{ContainerWebrootPath: "/var/www/html"}, "/deployments", "") + deployment := &models.Deployment{ + Name: "tenant-a", + Path: dir, + Metadata: &models.ServiceMetadata{ + Domains: []models.DomainConfig{ + {ID: "d1", Service: "api", ContainerPort: 8080, Domain: "api.example.com"}, + }, + }, + } + + config, err := m.generateMultiDomainConfig(deployment) + if err != nil { + t.Fatalf("generateMultiDomainConfig failed: %v", err) + } + + if !strings.Contains(config, "tenant-a-api:8080") { + t.Errorf("upstream should target the unique container name tenant-a-api, got:\n%s", config) + } + if strings.Contains(config, "set $upstream api:8080") { + t.Errorf("upstream must not use the bare service name 'api', got:\n%s", config) + } +} diff --git a/pkg/models/deployment.go b/pkg/models/deployment.go index f4abbb1..34a975e 100644 --- a/pkg/models/deployment.go +++ b/pkg/models/deployment.go @@ -20,17 +20,23 @@ type Service struct { Health string `json:"health,omitempty"` Ports []string `json:"ports,omitempty"` Networks []string `json:"networks,omitempty"` + IsPrimary bool `json:"is_primary"` CreatedAt time.Time `json:"created_at"` } type ServiceMetadata struct { - Name string `yaml:"name" json:"name"` - Type string `yaml:"type" json:"type"` + Name string `yaml:"name" json:"name"` + Type string `yaml:"type" json:"type"` + // PrimaryService is the user-pinned primary service. When set it overrides + // auto-detection for the default-domain upstream and is preserved across + // compose updates and re-discovery. + PrimaryService string `yaml:"primary_service,omitempty" json:"primary_service,omitempty"` Networking NetworkingConfig `yaml:"networking" json:"networking"` SSL SSLConfig `yaml:"ssl" json:"ssl"` HealthCheck HealthCheckConfig `yaml:"healthcheck" json:"healthcheck"` QuickActions []QuickAction `yaml:"quick_actions,omitempty" json:"quick_actions,omitempty"` Security *DeploymentSecurityConfig `yaml:"security,omitempty" json:"security,omitempty"` + AccessGroups *AccessGroupsConfig `yaml:"access_groups,omitempty" json:"access_groups,omitempty"` Backup *BackupSpec `yaml:"backup,omitempty" json:"backup,omitempty"` ProtectedMode *ProtectedModeConfig `yaml:"protected_mode,omitempty" json:"protected_mode,omitempty"` RequirePlan bool `yaml:"require_plan,omitempty" json:"require_plan,omitempty"` @@ -70,6 +76,15 @@ type DatabaseConfig struct { IsShared bool `yaml:"is_shared,omitempty" json:"is_shared,omitempty"` } +// EffectivePrimaryService returns the service treated as the deployment's primary: +// the user-pinned PrimaryService when set, otherwise the auto-detected routing service. +func (m *ServiceMetadata) EffectivePrimaryService() string { + if m.PrimaryService != "" { + return m.PrimaryService + } + return m.Networking.Service +} + func (m *ServiceMetadata) GetDomains() []DomainConfig { if len(m.Domains) > 0 { return m.Domains @@ -78,6 +93,9 @@ func (m *ServiceMetadata) GetDomains() []DomainConfig { return nil } service := m.Networking.Service + if m.PrimaryService != "" { + service = m.PrimaryService + } if service == "" { service = m.Name } @@ -231,6 +249,31 @@ type ProtectedPath struct { Enabled bool `yaml:"enabled" json:"enabled"` } +// AccessGroupsConfig is a per-deployment east-west / egress access policy (AWS +// security-group style: allow flows to peer deployments or external CIDRs, with a default +// egress stance). This is a scaffold: rules are persisted and surfaced in the UI, but +// enforcement is not yet wired (see pkg/plugins/accessgroups). It deliberately does not +// overlap the security module, which owns HTTP/ingress IP blocking and rate limiting. +type AccessGroupsConfig struct { + Enabled bool `yaml:"enabled" json:"enabled"` + // Egress is the default outbound stance: "allow-all" (default) or "deny-all". + Egress string `yaml:"egress,omitempty" json:"egress,omitempty"` + // Allow lists permitted flows: east-west grants to a peer deployment, or external + // egress to a CIDR, evaluated when Egress is deny-all. + Allow []AccessRule `yaml:"allow,omitempty" json:"allow,omitempty"` +} + +type AccessRule struct { + ID string `yaml:"id,omitempty" json:"id,omitempty"` + // To is the peer deployment name for an east-west rule; CIDR is the target for + // external egress. Exactly one is expected per rule. + To string `yaml:"to,omitempty" json:"to,omitempty"` + CIDR string `yaml:"cidr,omitempty" json:"cidr,omitempty"` + Port int `yaml:"port,omitempty" json:"port,omitempty"` + Protocol string `yaml:"protocol,omitempty" json:"protocol,omitempty"` + Description string `yaml:"description,omitempty" json:"description,omitempty"` +} + type DeploymentRateLimit struct { Path string `yaml:"path" json:"path"` Rate int `yaml:"rate" json:"rate"` diff --git a/pkg/models/deployment_test.go b/pkg/models/deployment_test.go index 0a2a291..7001964 100644 --- a/pkg/models/deployment_test.go +++ b/pkg/models/deployment_test.go @@ -143,6 +143,25 @@ func TestServiceMetadata_HasMultipleDatabases(t *testing.T) { } } +func TestServiceMetadata_EffectivePrimaryService(t *testing.T) { + tests := []struct { + name string + meta ServiceMetadata + want string + }{ + {"user pin wins", ServiceMetadata{PrimaryService: "api", Networking: NetworkingConfig{Service: "web"}}, "api"}, + {"falls back to networking service", ServiceMetadata{Networking: NetworkingConfig{Service: "web"}}, "web"}, + {"empty when neither set", ServiceMetadata{}, ""}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := tt.meta.EffectivePrimaryService(); got != tt.want { + t.Errorf("EffectivePrimaryService() = %q, want %q", got, tt.want) + } + }) + } +} + func TestServiceMetadata_GetDomains(t *testing.T) { tests := []struct { name string @@ -175,6 +194,21 @@ func TestServiceMetadata_GetDomains(t *testing.T) { want: 1, wantService: "web", }, + { + name: "user-pinned primary service overrides the networking service", + metadata: ServiceMetadata{ + Name: "myapp", + PrimaryService: "api", + Networking: NetworkingConfig{ + Expose: true, + Domain: "myapp.example.com", + Service: "web", + ContainerPort: 8080, + }, + }, + want: 1, + wantService: "api", + }, { name: "falls back to metadata name when service is empty", metadata: ServiceMetadata{ diff --git a/pkg/plugins/accessgroups/plugin.go b/pkg/plugins/accessgroups/plugin.go new file mode 100644 index 0000000..e579628 --- /dev/null +++ b/pkg/plugins/accessgroups/plugin.go @@ -0,0 +1,161 @@ +// Package accessgroups is a built-in FlatRun app (plugin) that models per-deployment +// east-west and egress access policy, AWS security-group style: allow flows to peer +// deployments or external CIDRs, with a default egress stance. Rules are stored in a +// deployment's service.yml under `access_groups`. +// +// This is a scaffold: Apply is a no-op. When enforcement lands it will operate at the +// Docker-network / L3-L4 layer (dedicated/internal networks, selective connect, or +// iptables) and must not write nginx config or IP blocks: HTTP ingress and rate limiting +// are owned by the security module. +package accessgroups + +import ( + "fmt" + "net/http" + + "github.com/flatrun/agent/pkg/models" + "github.com/flatrun/agent/pkg/plugins" + "github.com/gin-gonic/gin" +) + +const ( + EgressAllowAll = "allow-all" + EgressDenyAll = "deny-all" +) + +// Plugin is the built-in Access Groups app. It implements plugins.Plugin so it can be +// registered in the plugin registry and listed as an installed app. +type Plugin struct { + // policyFor loads a deployment's stored access-groups config so the plan endpoint can + // describe real rules. May be nil (then plan returns an empty policy). + policyFor func(deployment string) (*models.AccessGroupsConfig, error) +} + +func New(policyFor func(deployment string) (*models.AccessGroupsConfig, error)) *Plugin { + return &Plugin{policyFor: policyFor} +} + +func (p *Plugin) Info() plugins.PluginInfo { + return plugins.PluginInfo{ + Name: "access-groups", + Version: "0.1.0", + DisplayName: "Access Groups", + Description: "Control which deployments may reach each other and what they can reach (east-west and egress rules). Enforcement coming soon.", + Author: "FlatRun", + Type: plugins.TypeIntegration, + Category: "networking", + Enabled: true, + Capabilities: []string{"network_policy"}, + } +} + +func (p *Plugin) Initialize(config map[string]interface{}) error { return nil } +func (p *Plugin) Start() error { return nil } +func (p *Plugin) Stop() error { return nil } + +func (p *Plugin) GetCapabilities() []plugins.Capability { + return []plugins.Capability{"network_policy"} +} + +func (p *Plugin) GetWidgetData(deploymentName string) (interface{}, error) { + policy, _ := p.load(deploymentName) + count := 0 + enabled := false + if policy != nil { + count = len(policy.Allow) + enabled = policy.Enabled + } + return map[string]interface{}{ + "plugin": "access-groups", + "name": deploymentName, + "enabled": enabled, + "rules": count, + }, nil +} + +// RegisterRoutes exposes a read-only plan endpoint so the UI can preview the flows a +// deployment's access-groups config would permit (enforcement is not wired yet). +func (p *Plugin) RegisterRoutes(router *gin.RouterGroup) error { + router.GET("/deployments/:name/access-groups/plan", func(c *gin.Context) { + policy, err := p.load(c.Param("name")) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "deployment not found"}) + return + } + if err := Validate(policy); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"plan": Plan(policy), "enforced": false}) + }) + return nil +} + +func (p *Plugin) load(deployment string) (*models.AccessGroupsConfig, error) { + if p.policyFor == nil { + return nil, nil + } + return p.policyFor(deployment) +} + +// Validate checks that a policy is well formed before it is saved or applied. +func Validate(policy *models.AccessGroupsConfig) error { + if policy == nil { + return nil + } + switch policy.Egress { + case "", EgressAllowAll, EgressDenyAll: + default: + return fmt.Errorf("access_groups.egress must be %q or %q, got %q", EgressAllowAll, EgressDenyAll, policy.Egress) + } + for i, r := range policy.Allow { + if (r.To == "") == (r.CIDR == "") { + return fmt.Errorf("access_groups.allow[%d] must set exactly one of `to` (peer deployment) or `cidr` (external)", i) + } + if r.Port < 0 || r.Port > 65535 { + return fmt.Errorf("access_groups.allow[%d] port %d out of range", i, r.Port) + } + if r.Protocol != "" && r.Protocol != "tcp" && r.Protocol != "udp" { + return fmt.Errorf("access_groups.allow[%d] protocol must be tcp or udp, got %q", i, r.Protocol) + } + } + return nil +} + +// Plan returns a human-readable description of the flows the policy would permit, so the +// UI and tests can exercise the rule shape before real enforcement exists. +func Plan(policy *models.AccessGroupsConfig) []string { + if policy == nil || !policy.Enabled { + return nil + } + + egress := policy.Egress + if egress == "" { + egress = EgressAllowAll + } + plan := []string{fmt.Sprintf("default egress: %s", egress)} + + for _, r := range policy.Allow { + proto := r.Protocol + if proto == "" { + proto = "tcp" + } + target, kind := r.To, "deployment" + if target == "" { + target, kind = r.CIDR, "cidr" + } + port := "any" + if r.Port != 0 { + port = fmt.Sprintf("%d", r.Port) + } + plan = append(plan, fmt.Sprintf("allow %s/%s -> %s %s", proto, port, kind, target)) + } + return plan +} + +// Apply would reconcile host/Docker network state to match the policy. It is a no-op +// scaffold that only validates the policy; no network state is changed yet. +func Apply(deployment string, policy *models.AccessGroupsConfig) error { + // TODO: enforce east-west + egress rules via Docker internal networks / iptables. + return Validate(policy) +} diff --git a/pkg/plugins/accessgroups/plugin_test.go b/pkg/plugins/accessgroups/plugin_test.go new file mode 100644 index 0000000..7c38353 --- /dev/null +++ b/pkg/plugins/accessgroups/plugin_test.go @@ -0,0 +1,99 @@ +package accessgroups + +import ( + "strings" + "testing" + + "github.com/flatrun/agent/pkg/models" +) + +func TestValidate(t *testing.T) { + tests := []struct { + name string + policy *models.AccessGroupsConfig + wantErr string + }{ + {"nil is valid", nil, ""}, + {"empty egress ok", &models.AccessGroupsConfig{Enabled: true}, ""}, + { + "bad egress", + &models.AccessGroupsConfig{Egress: "block-everything"}, + "egress must be", + }, + { + "rule needs exactly one target", + &models.AccessGroupsConfig{Allow: []models.AccessRule{{To: "db", CIDR: "10.0.0.0/8"}}}, + "exactly one of", + }, + { + "rule with neither target", + &models.AccessGroupsConfig{Allow: []models.AccessRule{{Port: 5432}}}, + "exactly one of", + }, + { + "port out of range", + &models.AccessGroupsConfig{Allow: []models.AccessRule{{To: "db", Port: 70000}}}, + "out of range", + }, + { + "bad protocol", + &models.AccessGroupsConfig{Allow: []models.AccessRule{{To: "db", Protocol: "icmp"}}}, + "protocol must be", + }, + { + "valid deny-all with east-west allow", + &models.AccessGroupsConfig{Enabled: true, Egress: EgressDenyAll, Allow: []models.AccessRule{{To: "db", Port: 5432, Protocol: "tcp"}}}, + "", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := Validate(tt.policy) + if tt.wantErr == "" { + if err != nil { + t.Fatalf("Validate() = %v, want nil", err) + } + return + } + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("Validate() = %v, want error containing %q", err, tt.wantErr) + } + }) + } +} + +func TestPlan(t *testing.T) { + if Plan(nil) != nil { + t.Error("Plan(nil) should be nil") + } + if Plan(&models.AccessGroupsConfig{Enabled: false}) != nil { + t.Error("Plan(disabled) should be nil") + } + + policy := &models.AccessGroupsConfig{ + Enabled: true, + Egress: EgressDenyAll, + Allow: []models.AccessRule{ + {To: "db", Port: 5432}, + {CIDR: "10.0.0.0/8", Protocol: "udp", Port: 53}, + }, + } + plan := Plan(policy) + joined := strings.Join(plan, "\n") + if !strings.Contains(joined, "default egress: deny-all") { + t.Errorf("plan should state the default egress, got:\n%s", joined) + } + if !strings.Contains(joined, "allow tcp/5432 -> deployment db") { + t.Errorf("plan should describe the east-west allow with default tcp, got:\n%s", joined) + } + if !strings.Contains(joined, "allow udp/53 -> cidr 10.0.0.0/8") { + t.Errorf("plan should describe the egress CIDR allow, got:\n%s", joined) + } +} + +func TestInfoListsAsApp(t *testing.T) { + info := New(nil).Info() + if info.Name != "access-groups" || info.DisplayName != "Access Groups" { + t.Errorf("unexpected plugin info: %+v", info) + } +} diff --git a/templates/infra/nginx/nginx.conf b/templates/infra/nginx/nginx.conf index 5d35c79..93fca17 100644 --- a/templates/infra/nginx/nginx.conf +++ b/templates/infra/nginx/nginx.conf @@ -22,6 +22,7 @@ http { keepalive_timeout 65; types_hash_max_size 2048; server_names_hash_bucket_size 128; + server_names_hash_max_size 2048; # Gzip compression gzip on; diff --git a/templates/infra/nginx/nginx.lua.conf b/templates/infra/nginx/nginx.lua.conf index 9e23892..89cde95 100644 --- a/templates/infra/nginx/nginx.lua.conf +++ b/templates/infra/nginx/nginx.lua.conf @@ -21,6 +21,8 @@ http { tcp_nodelay on; keepalive_timeout 65; types_hash_max_size 2048; + server_names_hash_bucket_size 128; + server_names_hash_max_size 2048; # Gzip compression gzip on; From d0b117f23ae61d576f752d1584e91618c60ef7ef Mon Sep 17 00:00:00 2001 From: nfebe Date: Tue, 30 Jun 2026 14:33:09 +0100 Subject: [PATCH 2/4] feat(agent): Add an out-of-process plugin framework and unify built-in apps Plugins can now be standalone binaries that the agent launches as subprocesses and reverse-proxies, so an app ships its own API and UI without being compiled into the agent or run as a container. The plugin contract is split into small capability interfaces, so an app implements only what it needs: a firewall has no services to start or stop. The built-in apps, a host firewall and the DNS providers, now register through one registry and list uniformly alongside any installed plugins. The marketplace catalog is fetched through the agent rather than the browser, which removes the cross-origin restriction that blocked it; the upstream is configurable. Drops the per-deployment access-groups app, which only duplicated the network isolation Docker already provides between deployments by default. --- examples/plugins/hello/main.go | 29 ++++ internal/api/metadata_update_test.go | 19 --- internal/api/server.go | 147 ++++++++++++---- internal/pluginhost/host.go | 214 ++++++++++++++++++++++++ internal/pluginhost/host_test.go | 90 ++++++++++ pkg/models/deployment.go | 26 --- pkg/pluginapi/pluginapi.go | 29 ++++ pkg/plugins/accessgroups/plugin.go | 161 ------------------ pkg/plugins/accessgroups/plugin_test.go | 99 ----------- pkg/plugins/firewall/firewall.go | 165 ++++++++++++++++++ pkg/plugins/firewall/firewall_test.go | 93 ++++++++++ pkg/plugins/firewall/plugin.go | 82 +++++++++ pkg/plugins/registry.go | 31 +++- pkg/plugins/registry_test.go | 48 ++++++ pkg/plugins/types.go | 27 ++- pkg/pluginsdk/sdk.go | 67 ++++++++ 16 files changed, 986 insertions(+), 341 deletions(-) create mode 100644 examples/plugins/hello/main.go create mode 100644 internal/pluginhost/host.go create mode 100644 internal/pluginhost/host_test.go create mode 100644 pkg/pluginapi/pluginapi.go delete mode 100644 pkg/plugins/accessgroups/plugin.go delete mode 100644 pkg/plugins/accessgroups/plugin_test.go create mode 100644 pkg/plugins/firewall/firewall.go create mode 100644 pkg/plugins/firewall/firewall_test.go create mode 100644 pkg/plugins/firewall/plugin.go create mode 100644 pkg/plugins/registry_test.go create mode 100644 pkg/pluginsdk/sdk.go diff --git a/examples/plugins/hello/main.go b/examples/plugins/hello/main.go new file mode 100644 index 0000000..8fdf385 --- /dev/null +++ b/examples/plugins/hello/main.go @@ -0,0 +1,29 @@ +// Command hello is a sample FlatRun plugin: a standalone binary that serves an API the agent +// reverse-proxies under /api/v1/plugins/hello/. Build it with: +// +// go build -o plugin ./examples/plugins/hello +// +// and drop the resulting "plugin" binary in /.flatrun/plugins/hello/. +package main + +import ( + "net/http" + + "github.com/flatrun/agent/pkg/pluginapi" + "github.com/flatrun/agent/pkg/pluginsdk" +) + +func main() { + mux := http.NewServeMux() + mux.HandleFunc("/hello", func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"message":"hello from the flatrun plugin"}`)) + }) + + _ = pluginsdk.Serve(pluginapi.Info{ + Name: "hello", + Version: "0.1.0", + DisplayName: "Hello Plugin", + Description: "A sample out-of-process plugin.", + }, mux) +} diff --git a/internal/api/metadata_update_test.go b/internal/api/metadata_update_test.go index c430690..22e32e0 100644 --- a/internal/api/metadata_update_test.go +++ b/internal/api/metadata_update_test.go @@ -105,25 +105,6 @@ func TestMergeMetadata_PrimaryServiceSyncsRoutingService(t *testing.T) { } } -// An access_groups update is persisted through the metadata merge, while unsent fields -// are preserved. -func TestMergeMetadata_AccessGroups(t *testing.T) { - existing := &models.ServiceMetadata{Name: "shop", Type: "laravel"} - - sentFields, incoming := parseTestJSON(t, `{"access_groups": {"enabled": true, "egress": "deny-all", "allow": [{"to": "db", "port": 5432}]}}`) - merged := mergeMetadata(existing, &incoming, sentFields) - - if merged.AccessGroups == nil || !merged.AccessGroups.Enabled { - t.Fatalf("access_groups should be set and enabled, got %+v", merged.AccessGroups) - } - if merged.AccessGroups.Egress != "deny-all" || len(merged.AccessGroups.Allow) != 1 || merged.AccessGroups.Allow[0].To != "db" { - t.Errorf("access_groups not merged correctly: %+v", merged.AccessGroups) - } - if merged.Type != "laravel" { - t.Error("unsent field Type should be preserved") - } -} - func TestMergeMetadata_SentFieldOverwritesExisting(t *testing.T) { existing := &models.ServiceMetadata{ Name: "old-name", diff --git a/internal/api/server.go b/internal/api/server.go index 366847d..88539c4 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -10,6 +10,8 @@ import ( "log" "math/big" "net/http" + "net/http/httputil" + "net/url" "os" "os/exec" "path" @@ -37,6 +39,7 @@ import ( "github.com/flatrun/agent/internal/infra" "github.com/flatrun/agent/internal/networks" "github.com/flatrun/agent/internal/plan" + "github.com/flatrun/agent/internal/pluginhost" "github.com/flatrun/agent/internal/proxy" "github.com/flatrun/agent/internal/scheduler" "github.com/flatrun/agent/internal/security" @@ -47,8 +50,8 @@ import ( "github.com/flatrun/agent/pkg/config" "github.com/flatrun/agent/pkg/models" "github.com/flatrun/agent/pkg/plugins" - "github.com/flatrun/agent/pkg/plugins/accessgroups" dnsPlugins "github.com/flatrun/agent/pkg/plugins/dns" + "github.com/flatrun/agent/pkg/plugins/firewall" "github.com/flatrun/agent/pkg/subdomain" "github.com/flatrun/agent/pkg/version" "github.com/flatrun/agent/templates" @@ -67,7 +70,9 @@ type Server struct { certsDiscovery *certs.Discovery networksManager *networks.Manager pluginRegistry *plugins.Registry - accessGroups *accessgroups.Plugin + firewall *firewall.Plugin + builtinDNS []plugins.Plugin + pluginHost *pluginhost.Host authMiddleware *auth.Middleware authManager *auth.Manager proxyOrchestrator *proxy.Orchestrator @@ -158,19 +163,26 @@ func New(cfg *config.Config, configPath string) *Server { pluginRegistry := plugins.NewRegistry(pluginsDir) _ = pluginRegistry.LoadFromDisk() - // Access Groups is a built-in app: register it so it lists as an installed plugin and - // can serve its routes. Its plan endpoint reads the deployment's stored policy. - accessGroupsPlugin := accessgroups.New(func(name string) (*models.AccessGroupsConfig, error) { - dep, err := manager.GetDeployment(name) - if err != nil { - return nil, err - } - if dep.Metadata == nil { - return nil, nil - } - return dep.Metadata.AccessGroups, nil - }) - _ = pluginRegistry.Register(accessGroupsPlugin) + firewallPlugin := firewall.New(firewall.NewStore(cfg.DeploymentsPath)) + _ = pluginRegistry.Register(firewallPlugin) + + builtinDNS := []plugins.Plugin{ + dnsPlugins.NewCloudflarePlugin(), + dnsPlugins.NewRoute53Plugin(), + dnsPlugins.NewDigitalOceanPlugin(), + dnsPlugins.NewHetznerPlugin(), + } + for _, p := range builtinDNS { + _ = pluginRegistry.Register(p) + } + + agentURL := fmt.Sprintf("http://127.0.0.1:%d/api", cfg.API.Port) + pluginHost := pluginhost.New( + filepath.Join(cfg.DeploymentsPath, ".flatrun", "plugins"), + filepath.Join(cfg.DeploymentsPath, ".flatrun", "run"), + agentURL, + "", + ) authMiddleware := auth.NewMiddleware(&cfg.Auth) setupManager := setup.NewManager(cfg, configPath) @@ -285,7 +297,9 @@ func New(cfg *config.Config, configPath string) *Server { certsDiscovery: certsDiscovery, networksManager: networksManager, pluginRegistry: pluginRegistry, - accessGroups: accessGroupsPlugin, + firewall: firewallPlugin, + builtinDNS: builtinDNS, + pluginHost: pluginHost, authMiddleware: authMiddleware, authManager: authManager, proxyOrchestrator: proxyOrchestrator, @@ -475,6 +489,9 @@ func (s *Server) setupRoutes() { protected.GET("/plugins", s.authMiddleware.RequirePermission(auth.PermTemplatesRead), s.listPlugins) protected.GET("/plugins/:name", s.authMiddleware.RequirePermission(auth.PermTemplatesRead), s.getPlugin) protected.POST("/plugins/:name/deployments", s.authMiddleware.RequirePermission(auth.PermTemplatesWrite), s.createPluginDeployment) + + protected.Any("/plugin/:name/*proxyPath", s.authMiddleware.RequirePermission(auth.PermTemplatesRead), s.proxyToPlugin) + protected.Any("/marketplace/*path", s.authMiddleware.RequirePermission(auth.PermTemplatesRead), s.proxyMarketplace) protected.GET("/templates", s.authMiddleware.RequirePermission(auth.PermTemplatesRead), s.listTemplates) protected.GET("/templates/categories", s.authMiddleware.RequirePermission(auth.PermTemplatesRead), s.getTemplateCategories) protected.POST("/templates/refresh", s.authMiddleware.RequirePermission(auth.PermTemplatesWrite), s.refreshTemplates) @@ -693,18 +710,18 @@ func (s *Server) setupRoutes() { { dnsGroup.GET("/providers", s.listDNSProviders) - // Register DNS plugin routes - _ = dnsPlugins.NewCloudflarePlugin().RegisterRoutes(dnsGroup) - _ = dnsPlugins.NewRoute53Plugin().RegisterRoutes(dnsGroup) - _ = dnsPlugins.NewDigitalOceanPlugin().RegisterRoutes(dnsGroup) - _ = dnsPlugins.NewHetznerPlugin().RegisterRoutes(dnsGroup) + for _, p := range s.builtinDNS { + if rp, ok := p.(plugins.RoutablePlugin); ok { + _ = rp.RegisterRoutes(dnsGroup) + } + } // PowerDNS routes NewPowerDNSHandlers(s.powerDNSManager).RegisterRoutes(protected) } - // Access Groups built-in app routes (read-only plan; enforcement not wired yet) - _ = s.accessGroups.RegisterRoutes(protected) + // Firewall built-in app routes (config + plan; enforcement not wired yet) + _ = s.firewall.RegisterRoutes(protected) // Cluster endpoints clusterGroup := protected.Group("/cluster") @@ -742,15 +759,18 @@ func (s *Server) Start() error { Handler: s.router, } - // Refresh a managed base nginx config so an upgraded agent picks up template changes - // (e.g. server_names_hash sizing) without waiting on a security toggle. Best-effort and - // off the startup path since it may reload nginx once the container is ready. go func() { if err := s.infraManager.EnsureBaseNginxConfig(); err != nil { log.Printf("[infra] failed to refresh base nginx config on startup: %v", err) } }() + go func() { + if err := s.pluginHost.Start(); err != nil { + log.Printf("[pluginhost] failed to start plugins: %v", err) + } + }() + if s.config.Certbot.Enabled && s.config.Certbot.AutoRenewalEnabled { s.certRenewer = ssl.NewRenewer( s.proxyOrchestrator.SSLManager(), @@ -770,6 +790,9 @@ func (s *Server) Start() error { } func (s *Server) Stop() error { + if s.pluginHost != nil { + s.pluginHost.Stop() + } if s.certRenewer != nil { s.certRenewer.Stop() } @@ -1744,9 +1767,6 @@ func mergeMetadata(existing, incoming *models.ServiceMetadata, sentFields map[st if _, ok := sentFields["security"]; ok { merged.Security = incoming.Security } - if _, ok := sentFields["access_groups"]; ok { - merged.AccessGroups = incoming.AccessGroups - } if _, ok := sentFields["backup"]; ok { merged.Backup = incoming.Backup } @@ -2928,11 +2948,80 @@ func (s *Server) generateSubdomain(c *gin.Context) { func (s *Server) listPlugins(c *gin.Context) { pluginList := s.pluginRegistry.List() + for _, info := range s.pluginHost.Infos() { + pluginList = append(pluginList, plugins.PluginInfo{ + Name: info.Name, + Version: info.Version, + DisplayName: info.DisplayName, + Description: info.Description, + Capabilities: info.Capabilities, + Type: plugins.TypeIntegration, + Enabled: true, + }) + } + c.JSON(http.StatusOK, gin.H{ "plugins": pluginList, }) } +func marketplaceAPIBase() string { + if v := os.Getenv("FLATRUN_MARKETPLACE_API"); v != "" { + return v + } + return "https://api.flatrun.dev/api/v1" +} + +func (s *Server) proxyMarketplace(c *gin.Context) { + upstream, err := url.Parse(marketplaceAPIBase()) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "invalid marketplace upstream"}) + return + } + rel := c.Param("path") + proxy := &httputil.ReverseProxy{ + Director: func(req *http.Request) { + req.URL.Scheme = upstream.Scheme + req.URL.Host = upstream.Host + req.URL.Path = strings.TrimRight(upstream.Path, "/") + rel + req.Host = upstream.Host + req.Header.Del("Authorization") + req.Header.Del("Cookie") + req.Header.Del("X-Api-Key") + }, + // Drop the upstream's CORS headers; the agent sets its own, and two + // Access-Control-Allow-Origin headers make the browser reject the response. + ModifyResponse: func(resp *http.Response) error { + for _, h := range []string{ + "Access-Control-Allow-Origin", + "Access-Control-Allow-Credentials", + "Access-Control-Allow-Methods", + "Access-Control-Allow-Headers", + "Access-Control-Expose-Headers", + "Access-Control-Max-Age", + } { + resp.Header.Del(h) + } + return nil + }, + } + proxy.ServeHTTP(c.Writer, c.Request) +} + +func (s *Server) proxyToPlugin(c *gin.Context) { + name := c.Param("name") + proxy, ok := s.pluginHost.Proxy(name) + if !ok { + c.JSON(http.StatusNotFound, gin.H{"error": "plugin not running"}) + return + } + c.Request.URL.Path = c.Param("proxyPath") + if c.Request.URL.Path == "" { + c.Request.URL.Path = "/" + } + proxy.ServeHTTP(c.Writer, c.Request) +} + func (s *Server) getPlugin(c *gin.Context) { name := c.Param("name") diff --git a/internal/pluginhost/host.go b/internal/pluginhost/host.go new file mode 100644 index 0000000..940776e --- /dev/null +++ b/internal/pluginhost/host.go @@ -0,0 +1,214 @@ +// Package pluginhost launches out-of-process plugin binaries (//plugin), +// each listening on a private unix socket, and reverse-proxies their HTTP routes. +package pluginhost + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "log" + "net" + "net/http" + "net/http/httputil" + "os" + "os/exec" + "path/filepath" + "sync" + "syscall" + "time" + + "github.com/flatrun/agent/pkg/pluginapi" +) + +type managed struct { + info pluginapi.Info + socket string + cmd *exec.Cmd + proxy *httputil.ReverseProxy +} + +type Host struct { + pluginsDir string + runtimeDir string + agentURL string + token string + handshake string + + mu sync.RWMutex + running map[string]*managed +} + +func New(pluginsDir, runtimeDir, agentURL, token string) *Host { + return &Host{ + pluginsDir: pluginsDir, + runtimeDir: runtimeDir, + agentURL: agentURL, + token: token, + handshake: randomCookie(), + running: make(map[string]*managed), + } +} + +// Start launches every plugin binary; a plugin that fails to launch is logged and skipped. +func (h *Host) Start() error { + if h.pluginsDir == "" { + return nil + } + if err := os.MkdirAll(h.runtimeDir, 0755); err != nil { + return err + } + entries, err := os.ReadDir(h.pluginsDir) + if err != nil { + if os.IsNotExist(err) { + return nil + } + return err + } + for _, entry := range entries { + if !entry.IsDir() { + continue + } + bin := filepath.Join(h.pluginsDir, entry.Name(), "plugin") + if !isExecutable(bin) { + continue + } + if err := h.launch(entry.Name(), bin); err != nil { + log.Printf("[pluginhost] %s failed to start: %v", entry.Name(), err) + } + } + return nil +} + +func (h *Host) launch(name, bin string) error { + socket := filepath.Join(h.runtimeDir, name+".sock") + _ = os.Remove(socket) + + cmd := exec.Command(bin) + cmd.Env = append(os.Environ(), + pluginapi.EnvSocket+"="+socket, + pluginapi.EnvHandshake+"="+h.handshake, + pluginapi.EnvAgentURL+"="+h.agentURL, + pluginapi.EnvToken+"="+h.token, + ) + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + if err := cmd.Start(); err != nil { + return err + } + + info, err := h.awaitInfo(socket) + if err != nil { + _ = cmd.Process.Kill() + return err + } + + h.mu.Lock() + h.running[name] = &managed{info: info, socket: socket, cmd: cmd, proxy: newUnixProxy(socket)} + h.mu.Unlock() + log.Printf("[pluginhost] started %s (%s)", name, info.Version) + return nil +} + +func (h *Host) awaitInfo(socket string) (pluginapi.Info, error) { + client := &http.Client{Transport: unixTransport(socket), Timeout: 2 * time.Second} + deadline := time.Now().Add(5 * time.Second) + for { + info, err := fetchInfo(client, h.handshake) + if err == nil { + return info, nil + } + if time.Now().After(deadline) { + return pluginapi.Info{}, fmt.Errorf("plugin did not report info in time: %w", err) + } + time.Sleep(100 * time.Millisecond) + } +} + +func fetchInfo(client *http.Client, handshake string) (pluginapi.Info, error) { + resp, err := client.Get("http://plugin" + pluginapi.InfoPath) + if err != nil { + return pluginapi.Info{}, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return pluginapi.Info{}, fmt.Errorf("info returned %d", resp.StatusCode) + } + if resp.Header.Get(pluginapi.HandshakeHeader) != handshake { + return pluginapi.Info{}, fmt.Errorf("handshake mismatch") + } + var info pluginapi.Info + if err := json.NewDecoder(resp.Body).Decode(&info); err != nil { + return pluginapi.Info{}, err + } + return info, nil +} + +// Proxy returns a running plugin's reverse proxy. The caller must set the request path +// relative to the plugin (strip the /plugin/ prefix) before serving. +func (h *Host) Proxy(name string) (*httputil.ReverseProxy, bool) { + h.mu.RLock() + defer h.mu.RUnlock() + m, ok := h.running[name] + if !ok { + return nil, false + } + return m.proxy, true +} + +func (h *Host) Infos() []pluginapi.Info { + h.mu.RLock() + defer h.mu.RUnlock() + out := make([]pluginapi.Info, 0, len(h.running)) + for _, m := range h.running { + out = append(out, m.info) + } + return out +} + +func (h *Host) Stop() { + h.mu.Lock() + defer h.mu.Unlock() + for name, m := range h.running { + if m.cmd.Process != nil { + _ = m.cmd.Process.Signal(syscall.SIGTERM) + } + _ = os.Remove(m.socket) + delete(h.running, name) + } +} + +func unixTransport(socket string) *http.Transport { + return &http.Transport{ + DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, "unix", socket) + }, + } +} + +func newUnixProxy(socket string) *httputil.ReverseProxy { + return &httputil.ReverseProxy{ + Director: func(req *http.Request) { + req.URL.Scheme = "http" + req.URL.Host = "plugin" + }, + Transport: unixTransport(socket), + } +} + +func isExecutable(path string) bool { + info, err := os.Stat(path) + if err != nil || info.IsDir() { + return false + } + return info.Mode().Perm()&0111 != 0 +} + +func randomCookie() string { + b := make([]byte, 16) + if _, err := rand.Read(b); err != nil { + return "flatrun-plugin-handshake" + } + return hex.EncodeToString(b) +} diff --git a/internal/pluginhost/host_test.go b/internal/pluginhost/host_test.go new file mode 100644 index 0000000..690dee5 --- /dev/null +++ b/internal/pluginhost/host_test.go @@ -0,0 +1,90 @@ +package pluginhost + +import ( + "net" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" +) + +// The unix reverse proxy forwards a request to a plugin listening on its socket. +func TestUnixProxyForwards(t *testing.T) { + dir := t.TempDir() + socket := filepath.Join(dir, "p.sock") + + ln, err := net.Listen("unix", socket) + if err != nil { + t.Fatal(err) + } + srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/hello" { + _, _ = w.Write([]byte("hi from plugin")) + return + } + w.WriteHeader(http.StatusNotFound) + })} + go func() { _ = srv.Serve(ln) }() + defer srv.Close() + + proxy := newUnixProxy(socket) + rec := httptest.NewRecorder() + proxy.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/hello", nil)) + + if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), "hi from plugin") { + t.Fatalf("proxy did not forward to the plugin: code=%d body=%q", rec.Code, rec.Body.String()) + } +} + +// End to end: build the sample plugin binary, launch it through the host, and proxy a request. +func TestHostLaunchesAndProxiesBinaryPlugin(t *testing.T) { + if testing.Short() { + t.Skip("skipping subprocess build in short mode") + } + goBin, err := exec.LookPath("go") + if err != nil { + t.Skip("go toolchain not available") + } + + base := t.TempDir() + helloDir := filepath.Join(base, "plugins", "hello") + if err := os.MkdirAll(helloDir, 0755); err != nil { + t.Fatal(err) + } + + build := exec.Command(goBin, "build", "-o", filepath.Join(helloDir, "plugin"), "github.com/flatrun/agent/examples/plugins/hello") + if out, err := build.CombinedOutput(); err != nil { + t.Fatalf("building sample plugin: %v\n%s", err, out) + } + + h := New(filepath.Join(base, "plugins"), filepath.Join(base, "run"), "", "") + if err := h.Start(); err != nil { + t.Fatal(err) + } + defer h.Stop() + + if len(h.Infos()) != 1 || h.Infos()[0].Name != "hello" { + t.Fatalf("expected the hello plugin to be running, got %+v", h.Infos()) + } + + proxy, ok := h.Proxy("hello") + if !ok { + t.Fatal("hello plugin proxy not found") + } + rec := httptest.NewRecorder() + proxy.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/hello", nil)) + if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), "hello from the flatrun plugin") { + t.Fatalf("proxied plugin response = code %d body %q", rec.Code, rec.Body.String()) + } + + // Stop terminates the subprocess and clears the socket. + h.Stop() + time.Sleep(50 * time.Millisecond) + if _, ok := h.Proxy("hello"); ok { + t.Error("plugin should be gone after Stop") + } +} diff --git a/pkg/models/deployment.go b/pkg/models/deployment.go index 34a975e..c937ec4 100644 --- a/pkg/models/deployment.go +++ b/pkg/models/deployment.go @@ -36,7 +36,6 @@ type ServiceMetadata struct { HealthCheck HealthCheckConfig `yaml:"healthcheck" json:"healthcheck"` QuickActions []QuickAction `yaml:"quick_actions,omitempty" json:"quick_actions,omitempty"` Security *DeploymentSecurityConfig `yaml:"security,omitempty" json:"security,omitempty"` - AccessGroups *AccessGroupsConfig `yaml:"access_groups,omitempty" json:"access_groups,omitempty"` Backup *BackupSpec `yaml:"backup,omitempty" json:"backup,omitempty"` ProtectedMode *ProtectedModeConfig `yaml:"protected_mode,omitempty" json:"protected_mode,omitempty"` RequirePlan bool `yaml:"require_plan,omitempty" json:"require_plan,omitempty"` @@ -249,31 +248,6 @@ type ProtectedPath struct { Enabled bool `yaml:"enabled" json:"enabled"` } -// AccessGroupsConfig is a per-deployment east-west / egress access policy (AWS -// security-group style: allow flows to peer deployments or external CIDRs, with a default -// egress stance). This is a scaffold: rules are persisted and surfaced in the UI, but -// enforcement is not yet wired (see pkg/plugins/accessgroups). It deliberately does not -// overlap the security module, which owns HTTP/ingress IP blocking and rate limiting. -type AccessGroupsConfig struct { - Enabled bool `yaml:"enabled" json:"enabled"` - // Egress is the default outbound stance: "allow-all" (default) or "deny-all". - Egress string `yaml:"egress,omitempty" json:"egress,omitempty"` - // Allow lists permitted flows: east-west grants to a peer deployment, or external - // egress to a CIDR, evaluated when Egress is deny-all. - Allow []AccessRule `yaml:"allow,omitempty" json:"allow,omitempty"` -} - -type AccessRule struct { - ID string `yaml:"id,omitempty" json:"id,omitempty"` - // To is the peer deployment name for an east-west rule; CIDR is the target for - // external egress. Exactly one is expected per rule. - To string `yaml:"to,omitempty" json:"to,omitempty"` - CIDR string `yaml:"cidr,omitempty" json:"cidr,omitempty"` - Port int `yaml:"port,omitempty" json:"port,omitempty"` - Protocol string `yaml:"protocol,omitempty" json:"protocol,omitempty"` - Description string `yaml:"description,omitempty" json:"description,omitempty"` -} - type DeploymentRateLimit struct { Path string `yaml:"path" json:"path"` Rate int `yaml:"rate" json:"rate"` diff --git a/pkg/pluginapi/pluginapi.go b/pkg/pluginapi/pluginapi.go new file mode 100644 index 0000000..c173a32 --- /dev/null +++ b/pkg/pluginapi/pluginapi.go @@ -0,0 +1,29 @@ +// Package pluginapi is the dependency-free wire contract between the agent and a plugin +// binary: the plugin serves /_plugin/info (echoing the handshake) plus its own routes on the +// host-provided unix socket. +package pluginapi + +const ( + // EnvSocket is the unix socket path the plugin must listen on. + EnvSocket = "FLATRUN_PLUGIN_SOCKET" + // EnvHandshake is a per-launch cookie the plugin echoes so the host knows it spoke to + // the process it actually started, not something else bound to the socket. + EnvHandshake = "FLATRUN_PLUGIN_HANDSHAKE" + // EnvAgentURL and EnvToken let a plugin call back into the agent API. + EnvAgentURL = "FLATRUN_AGENT_URL" + EnvToken = "FLATRUN_PLUGIN_TOKEN" + + // InfoPath is the well-known endpoint every plugin serves. + InfoPath = "/_plugin/info" + // HandshakeHeader carries the echoed handshake cookie on the info response. + HandshakeHeader = "X-Flatrun-Plugin-Handshake" +) + +// Info is a plugin's self-reported identity and the capabilities it requests from the host. +type Info struct { + Name string `json:"name"` + Version string `json:"version"` + DisplayName string `json:"display_name"` + Description string `json:"description"` + Capabilities []string `json:"capabilities,omitempty"` +} diff --git a/pkg/plugins/accessgroups/plugin.go b/pkg/plugins/accessgroups/plugin.go deleted file mode 100644 index e579628..0000000 --- a/pkg/plugins/accessgroups/plugin.go +++ /dev/null @@ -1,161 +0,0 @@ -// Package accessgroups is a built-in FlatRun app (plugin) that models per-deployment -// east-west and egress access policy, AWS security-group style: allow flows to peer -// deployments or external CIDRs, with a default egress stance. Rules are stored in a -// deployment's service.yml under `access_groups`. -// -// This is a scaffold: Apply is a no-op. When enforcement lands it will operate at the -// Docker-network / L3-L4 layer (dedicated/internal networks, selective connect, or -// iptables) and must not write nginx config or IP blocks: HTTP ingress and rate limiting -// are owned by the security module. -package accessgroups - -import ( - "fmt" - "net/http" - - "github.com/flatrun/agent/pkg/models" - "github.com/flatrun/agent/pkg/plugins" - "github.com/gin-gonic/gin" -) - -const ( - EgressAllowAll = "allow-all" - EgressDenyAll = "deny-all" -) - -// Plugin is the built-in Access Groups app. It implements plugins.Plugin so it can be -// registered in the plugin registry and listed as an installed app. -type Plugin struct { - // policyFor loads a deployment's stored access-groups config so the plan endpoint can - // describe real rules. May be nil (then plan returns an empty policy). - policyFor func(deployment string) (*models.AccessGroupsConfig, error) -} - -func New(policyFor func(deployment string) (*models.AccessGroupsConfig, error)) *Plugin { - return &Plugin{policyFor: policyFor} -} - -func (p *Plugin) Info() plugins.PluginInfo { - return plugins.PluginInfo{ - Name: "access-groups", - Version: "0.1.0", - DisplayName: "Access Groups", - Description: "Control which deployments may reach each other and what they can reach (east-west and egress rules). Enforcement coming soon.", - Author: "FlatRun", - Type: plugins.TypeIntegration, - Category: "networking", - Enabled: true, - Capabilities: []string{"network_policy"}, - } -} - -func (p *Plugin) Initialize(config map[string]interface{}) error { return nil } -func (p *Plugin) Start() error { return nil } -func (p *Plugin) Stop() error { return nil } - -func (p *Plugin) GetCapabilities() []plugins.Capability { - return []plugins.Capability{"network_policy"} -} - -func (p *Plugin) GetWidgetData(deploymentName string) (interface{}, error) { - policy, _ := p.load(deploymentName) - count := 0 - enabled := false - if policy != nil { - count = len(policy.Allow) - enabled = policy.Enabled - } - return map[string]interface{}{ - "plugin": "access-groups", - "name": deploymentName, - "enabled": enabled, - "rules": count, - }, nil -} - -// RegisterRoutes exposes a read-only plan endpoint so the UI can preview the flows a -// deployment's access-groups config would permit (enforcement is not wired yet). -func (p *Plugin) RegisterRoutes(router *gin.RouterGroup) error { - router.GET("/deployments/:name/access-groups/plan", func(c *gin.Context) { - policy, err := p.load(c.Param("name")) - if err != nil { - c.JSON(http.StatusNotFound, gin.H{"error": "deployment not found"}) - return - } - if err := Validate(policy); err != nil { - c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) - return - } - c.JSON(http.StatusOK, gin.H{"plan": Plan(policy), "enforced": false}) - }) - return nil -} - -func (p *Plugin) load(deployment string) (*models.AccessGroupsConfig, error) { - if p.policyFor == nil { - return nil, nil - } - return p.policyFor(deployment) -} - -// Validate checks that a policy is well formed before it is saved or applied. -func Validate(policy *models.AccessGroupsConfig) error { - if policy == nil { - return nil - } - switch policy.Egress { - case "", EgressAllowAll, EgressDenyAll: - default: - return fmt.Errorf("access_groups.egress must be %q or %q, got %q", EgressAllowAll, EgressDenyAll, policy.Egress) - } - for i, r := range policy.Allow { - if (r.To == "") == (r.CIDR == "") { - return fmt.Errorf("access_groups.allow[%d] must set exactly one of `to` (peer deployment) or `cidr` (external)", i) - } - if r.Port < 0 || r.Port > 65535 { - return fmt.Errorf("access_groups.allow[%d] port %d out of range", i, r.Port) - } - if r.Protocol != "" && r.Protocol != "tcp" && r.Protocol != "udp" { - return fmt.Errorf("access_groups.allow[%d] protocol must be tcp or udp, got %q", i, r.Protocol) - } - } - return nil -} - -// Plan returns a human-readable description of the flows the policy would permit, so the -// UI and tests can exercise the rule shape before real enforcement exists. -func Plan(policy *models.AccessGroupsConfig) []string { - if policy == nil || !policy.Enabled { - return nil - } - - egress := policy.Egress - if egress == "" { - egress = EgressAllowAll - } - plan := []string{fmt.Sprintf("default egress: %s", egress)} - - for _, r := range policy.Allow { - proto := r.Protocol - if proto == "" { - proto = "tcp" - } - target, kind := r.To, "deployment" - if target == "" { - target, kind = r.CIDR, "cidr" - } - port := "any" - if r.Port != 0 { - port = fmt.Sprintf("%d", r.Port) - } - plan = append(plan, fmt.Sprintf("allow %s/%s -> %s %s", proto, port, kind, target)) - } - return plan -} - -// Apply would reconcile host/Docker network state to match the policy. It is a no-op -// scaffold that only validates the policy; no network state is changed yet. -func Apply(deployment string, policy *models.AccessGroupsConfig) error { - // TODO: enforce east-west + egress rules via Docker internal networks / iptables. - return Validate(policy) -} diff --git a/pkg/plugins/accessgroups/plugin_test.go b/pkg/plugins/accessgroups/plugin_test.go deleted file mode 100644 index 7c38353..0000000 --- a/pkg/plugins/accessgroups/plugin_test.go +++ /dev/null @@ -1,99 +0,0 @@ -package accessgroups - -import ( - "strings" - "testing" - - "github.com/flatrun/agent/pkg/models" -) - -func TestValidate(t *testing.T) { - tests := []struct { - name string - policy *models.AccessGroupsConfig - wantErr string - }{ - {"nil is valid", nil, ""}, - {"empty egress ok", &models.AccessGroupsConfig{Enabled: true}, ""}, - { - "bad egress", - &models.AccessGroupsConfig{Egress: "block-everything"}, - "egress must be", - }, - { - "rule needs exactly one target", - &models.AccessGroupsConfig{Allow: []models.AccessRule{{To: "db", CIDR: "10.0.0.0/8"}}}, - "exactly one of", - }, - { - "rule with neither target", - &models.AccessGroupsConfig{Allow: []models.AccessRule{{Port: 5432}}}, - "exactly one of", - }, - { - "port out of range", - &models.AccessGroupsConfig{Allow: []models.AccessRule{{To: "db", Port: 70000}}}, - "out of range", - }, - { - "bad protocol", - &models.AccessGroupsConfig{Allow: []models.AccessRule{{To: "db", Protocol: "icmp"}}}, - "protocol must be", - }, - { - "valid deny-all with east-west allow", - &models.AccessGroupsConfig{Enabled: true, Egress: EgressDenyAll, Allow: []models.AccessRule{{To: "db", Port: 5432, Protocol: "tcp"}}}, - "", - }, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - err := Validate(tt.policy) - if tt.wantErr == "" { - if err != nil { - t.Fatalf("Validate() = %v, want nil", err) - } - return - } - if err == nil || !strings.Contains(err.Error(), tt.wantErr) { - t.Fatalf("Validate() = %v, want error containing %q", err, tt.wantErr) - } - }) - } -} - -func TestPlan(t *testing.T) { - if Plan(nil) != nil { - t.Error("Plan(nil) should be nil") - } - if Plan(&models.AccessGroupsConfig{Enabled: false}) != nil { - t.Error("Plan(disabled) should be nil") - } - - policy := &models.AccessGroupsConfig{ - Enabled: true, - Egress: EgressDenyAll, - Allow: []models.AccessRule{ - {To: "db", Port: 5432}, - {CIDR: "10.0.0.0/8", Protocol: "udp", Port: 53}, - }, - } - plan := Plan(policy) - joined := strings.Join(plan, "\n") - if !strings.Contains(joined, "default egress: deny-all") { - t.Errorf("plan should state the default egress, got:\n%s", joined) - } - if !strings.Contains(joined, "allow tcp/5432 -> deployment db") { - t.Errorf("plan should describe the east-west allow with default tcp, got:\n%s", joined) - } - if !strings.Contains(joined, "allow udp/53 -> cidr 10.0.0.0/8") { - t.Errorf("plan should describe the egress CIDR allow, got:\n%s", joined) - } -} - -func TestInfoListsAsApp(t *testing.T) { - info := New(nil).Info() - if info.Name != "access-groups" || info.DisplayName != "Access Groups" { - t.Errorf("unexpected plugin info: %+v", info) - } -} diff --git a/pkg/plugins/firewall/firewall.go b/pkg/plugins/firewall/firewall.go new file mode 100644 index 0000000..b6a577e --- /dev/null +++ b/pkg/plugins/firewall/firewall.go @@ -0,0 +1,165 @@ +// Package firewall is a built-in FlatRun app (plugin) that models a host-wide firewall: +// default inbound/outbound stance plus allow/deny rules for traffic to and from the whole +// server. It operates at a different layer than the security module, which controls HTTP +// ingress at the proxy. +// +// This is a scaffold: rules are persisted, validated, and previewable, but Apply is a no-op. +// Enforcement will translate rules to nftables/iptables; until then nothing touches the host +// firewall, which avoids the obvious foot-gun of locking the operator out over SSH. +package firewall + +import ( + "fmt" + "net" + "os" + "path/filepath" + + "gopkg.in/yaml.v3" +) + +const ( + PolicyAllow = "allow" + PolicyDeny = "deny" + + DirInbound = "inbound" + DirOutbound = "outbound" +) + +// Config is the host firewall policy, stored globally in .flatrun/firewall.yml. +type Config struct { + Enabled bool `yaml:"enabled" json:"enabled"` + // DefaultInbound / DefaultOutbound are the stance applied to traffic not matched by a + // rule: "allow" (default) or "deny". + DefaultInbound string `yaml:"default_inbound,omitempty" json:"default_inbound,omitempty"` + DefaultOutbound string `yaml:"default_outbound,omitempty" json:"default_outbound,omitempty"` + Rules []Rule `yaml:"rules,omitempty" json:"rules,omitempty"` +} + +type Rule struct { + ID string `yaml:"id,omitempty" json:"id,omitempty"` + Direction string `yaml:"direction" json:"direction"` // inbound | outbound + Action string `yaml:"action" json:"action"` // allow | deny + Protocol string `yaml:"protocol,omitempty" json:"protocol,omitempty"` + Port int `yaml:"port,omitempty" json:"port,omitempty"` + // CIDR is the source for inbound rules or the destination for outbound rules. + CIDR string `yaml:"cidr,omitempty" json:"cidr,omitempty"` + Description string `yaml:"description,omitempty" json:"description,omitempty"` +} + +// Store loads and saves the host firewall config from a flat file. +type Store struct { + path string +} + +func NewStore(basePath string) *Store { + return &Store{path: filepath.Join(basePath, ".flatrun", "firewall.yml")} +} + +// Load returns the stored config, or a disabled default when no file exists yet. +func (s *Store) Load() (*Config, error) { + data, err := os.ReadFile(s.path) + if err != nil { + if os.IsNotExist(err) { + return &Config{}, nil + } + return nil, err + } + var cfg Config + if err := yaml.Unmarshal(data, &cfg); err != nil { + return nil, fmt.Errorf("invalid firewall config: %w", err) + } + return &cfg, nil +} + +func (s *Store) Save(cfg *Config) error { + if err := Validate(cfg); err != nil { + return err + } + if err := os.MkdirAll(filepath.Dir(s.path), 0755); err != nil { + return err + } + data, err := yaml.Marshal(cfg) + if err != nil { + return err + } + return os.WriteFile(s.path, data, 0644) +} + +func validPolicy(p string) bool { return p == "" || p == PolicyAllow || p == PolicyDeny } + +// Validate checks the firewall config is well formed before it is saved or applied. +func Validate(cfg *Config) error { + if cfg == nil { + return nil + } + if !validPolicy(cfg.DefaultInbound) || !validPolicy(cfg.DefaultOutbound) { + return fmt.Errorf("firewall default policy must be %q or %q", PolicyAllow, PolicyDeny) + } + for i, r := range cfg.Rules { + if r.Direction != DirInbound && r.Direction != DirOutbound { + return fmt.Errorf("firewall.rules[%d] direction must be %q or %q", i, DirInbound, DirOutbound) + } + if r.Action != PolicyAllow && r.Action != PolicyDeny { + return fmt.Errorf("firewall.rules[%d] action must be %q or %q", i, PolicyAllow, PolicyDeny) + } + if r.Protocol != "" && r.Protocol != "tcp" && r.Protocol != "udp" && r.Protocol != "any" { + return fmt.Errorf("firewall.rules[%d] protocol must be tcp, udp, or any", i) + } + if r.Port < 0 || r.Port > 65535 { + return fmt.Errorf("firewall.rules[%d] port %d out of range", i, r.Port) + } + if r.CIDR != "" { + if _, _, err := net.ParseCIDR(r.CIDR); err != nil { + return fmt.Errorf("firewall.rules[%d] cidr %q is not valid: %w", i, r.CIDR, err) + } + } + } + return nil +} + +// Plan returns a human-readable description of the rules that would be enforced, so the UI +// and tests can exercise the shape before real enforcement exists. +func Plan(cfg *Config) []string { + if cfg == nil || !cfg.Enabled { + return nil + } + inbound, outbound := cfg.DefaultInbound, cfg.DefaultOutbound + if inbound == "" { + inbound = PolicyAllow + } + if outbound == "" { + outbound = PolicyAllow + } + plan := []string{ + fmt.Sprintf("default inbound: %s", inbound), + fmt.Sprintf("default outbound: %s", outbound), + } + for _, r := range cfg.Rules { + proto := r.Protocol + if proto == "" { + proto = "any" + } + port := "any" + if r.Port != 0 { + port = fmt.Sprintf("%d", r.Port) + } + peer := r.CIDR + if peer == "" { + peer = "0.0.0.0/0" + } + prep := "from" + if r.Direction == DirOutbound { + prep = "to" + } + plan = append(plan, fmt.Sprintf("%s %s %s/%s %s %s", r.Action, r.Direction, proto, port, prep, peer)) + } + return plan +} + +// Apply would translate the config to host firewall rules. It is a no-op scaffold that only +// validates the config; the host firewall is never touched yet. +func Apply(cfg *Config) error { + // TODO: render rules to nftables/iptables with a safety net that preserves the active + // SSH session before committing a default-deny inbound policy. + return Validate(cfg) +} diff --git a/pkg/plugins/firewall/firewall_test.go b/pkg/plugins/firewall/firewall_test.go new file mode 100644 index 0000000..39cf71e --- /dev/null +++ b/pkg/plugins/firewall/firewall_test.go @@ -0,0 +1,93 @@ +package firewall + +import ( + "path/filepath" + "strings" + "testing" +) + +func TestValidate(t *testing.T) { + tests := []struct { + name string + cfg *Config + wantErr string + }{ + {"nil ok", nil, ""}, + {"empty ok", &Config{Enabled: true}, ""}, + {"bad default policy", &Config{DefaultInbound: "drop"}, "default policy"}, + {"bad direction", &Config{Rules: []Rule{{Direction: "sideways", Action: PolicyAllow}}}, "direction"}, + {"bad action", &Config{Rules: []Rule{{Direction: DirInbound, Action: "log"}}}, "action"}, + {"bad protocol", &Config{Rules: []Rule{{Direction: DirInbound, Action: PolicyAllow, Protocol: "icmp"}}}, "protocol"}, + {"bad port", &Config{Rules: []Rule{{Direction: DirInbound, Action: PolicyAllow, Port: 99999}}}, "out of range"}, + {"bad cidr", &Config{Rules: []Rule{{Direction: DirInbound, Action: PolicyAllow, CIDR: "not-a-cidr"}}}, "cidr"}, + { + "valid", + &Config{Enabled: true, DefaultInbound: PolicyDeny, Rules: []Rule{{Direction: DirInbound, Action: PolicyAllow, Protocol: "tcp", Port: 22, CIDR: "0.0.0.0/0"}}}, + "", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := Validate(tt.cfg) + if tt.wantErr == "" { + if err != nil { + t.Fatalf("Validate() = %v, want nil", err) + } + return + } + if err == nil || !strings.Contains(err.Error(), tt.wantErr) { + t.Fatalf("Validate() = %v, want error containing %q", err, tt.wantErr) + } + }) + } +} + +func TestPlan(t *testing.T) { + if Plan(&Config{Enabled: false}) != nil { + t.Error("Plan(disabled) should be nil") + } + cfg := &Config{ + Enabled: true, + DefaultInbound: PolicyDeny, + Rules: []Rule{ + {Direction: DirInbound, Action: PolicyAllow, Protocol: "tcp", Port: 22, CIDR: "0.0.0.0/0"}, + {Direction: DirOutbound, Action: PolicyDeny, CIDR: "10.0.0.0/8"}, + }, + } + joined := strings.Join(Plan(cfg), "\n") + if !strings.Contains(joined, "default inbound: deny") || !strings.Contains(joined, "default outbound: allow") { + t.Errorf("plan should state both default policies, got:\n%s", joined) + } + if !strings.Contains(joined, "allow inbound tcp/22 from 0.0.0.0/0") { + t.Errorf("plan should describe the inbound allow, got:\n%s", joined) + } + if !strings.Contains(joined, "deny outbound any/any to 10.0.0.0/8") { + t.Errorf("plan should describe the outbound deny, got:\n%s", joined) + } +} + +func TestStoreRoundtrip(t *testing.T) { + base := t.TempDir() + store := NewStore(base) + + // Loading before anything is saved returns a disabled default, not an error. + cfg, err := store.Load() + if err != nil || cfg.Enabled { + t.Fatalf("Load() before save = (%+v, %v), want disabled default", cfg, err) + } + + want := &Config{Enabled: true, DefaultInbound: PolicyDeny, Rules: []Rule{{Direction: DirInbound, Action: PolicyAllow, Port: 443}}} + if err := store.Save(want); err != nil { + t.Fatalf("Save() = %v", err) + } + if _, err := store.Load(); err != nil { + t.Fatalf("Load() after save = %v", err) + } + got, _ := store.Load() + if !got.Enabled || got.DefaultInbound != PolicyDeny || len(got.Rules) != 1 || got.Rules[0].Port != 443 { + t.Errorf("round-tripped config = %+v, want %+v", got, want) + } + if filepath.Base(store.path) != "firewall.yml" { + t.Errorf("unexpected store path %q", store.path) + } +} diff --git a/pkg/plugins/firewall/plugin.go b/pkg/plugins/firewall/plugin.go new file mode 100644 index 0000000..9378f1e --- /dev/null +++ b/pkg/plugins/firewall/plugin.go @@ -0,0 +1,82 @@ +package firewall + +import ( + "net/http" + + "github.com/flatrun/agent/pkg/plugins" + "github.com/gin-gonic/gin" +) + +// Plugin is the built-in Firewall app. It implements plugins.Plugin so it registers in the +// plugin registry and lists as an installed app. +type Plugin struct { + store *Store +} + +func New(store *Store) *Plugin { return &Plugin{store: store} } + +func (p *Plugin) Info() plugins.PluginInfo { + return plugins.PluginInfo{ + Name: "firewall", + Version: "0.1.0", + DisplayName: "Firewall", + Description: "Set the server's inbound and outbound traffic rules in one place. Enforcement coming soon.", + Author: "FlatRun", + Type: plugins.TypeIntegration, + Category: "security", + Enabled: true, + Capabilities: []string{"host_firewall"}, + } +} + +func (p *Plugin) GetCapabilities() []plugins.Capability { + return []plugins.Capability{"host_firewall"} +} + +func (p *Plugin) GetWidgetData(deploymentName string) (interface{}, error) { + cfg, err := p.store.Load() + if err != nil { + return nil, err + } + return map[string]interface{}{ + "plugin": "firewall", + "enabled": cfg.Enabled, + "rules": len(cfg.Rules), + }, nil +} + +// RegisterRoutes exposes the host firewall config and a read-only plan. Saving validates the +// config but does not yet enforce it. +func (p *Plugin) RegisterRoutes(router *gin.RouterGroup) error { + router.GET("/firewall", func(c *gin.Context) { + cfg, err := p.store.Load() + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, cfg) + }) + + router.PUT("/firewall", func(c *gin.Context) { + var cfg Config + if err := c.ShouldBindJSON(&cfg); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request body: " + err.Error()}) + return + } + if err := p.store.Save(&cfg); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"message": "firewall config saved", "enforced": false}) + }) + + router.GET("/firewall/plan", func(c *gin.Context) { + cfg, err := p.store.Load() + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"plan": Plan(cfg), "enforced": false}) + }) + return nil +} diff --git a/pkg/plugins/registry.go b/pkg/plugins/registry.go index b1cb6d8..064628e 100644 --- a/pkg/plugins/registry.go +++ b/pkg/plugins/registry.go @@ -40,8 +40,10 @@ func (r *Registry) Unregister(name string) error { defer r.mu.Unlock() if plugin, exists := r.plugins[name]; exists { - if err := plugin.Stop(); err != nil { - return err + if lc, ok := plugin.(LifecyclePlugin); ok { + if err := lc.Stop(); err != nil { + return err + } } delete(r.plugins, name) return nil @@ -50,6 +52,19 @@ func (r *Registry) Unregister(name string) error { return fmt.Errorf("plugin %s not found", name) } +// Plugins returns every registered plugin so the host can wire routes or inspect +// capabilities through the optional interfaces. +func (r *Registry) Plugins() []Plugin { + r.mu.RLock() + defer r.mu.RUnlock() + + out := make([]Plugin, 0, len(r.plugins)) + for _, plugin := range r.plugins { + out = append(out, plugin) + } + return out +} + func (r *Registry) Get(name string) (Plugin, bool) { r.mu.RLock() defer r.mu.RUnlock() @@ -127,8 +142,10 @@ func (r *Registry) StartAll() error { defer r.mu.RUnlock() for name, plugin := range r.plugins { - if err := plugin.Start(); err != nil { - return fmt.Errorf("failed to start plugin %s: %w", name, err) + if lc, ok := plugin.(LifecyclePlugin); ok { + if err := lc.Start(); err != nil { + return fmt.Errorf("failed to start plugin %s: %w", name, err) + } } } return nil @@ -139,8 +156,10 @@ func (r *Registry) StopAll() error { defer r.mu.RUnlock() for name, plugin := range r.plugins { - if err := plugin.Stop(); err != nil { - return fmt.Errorf("failed to stop plugin %s: %w", name, err) + if lc, ok := plugin.(LifecyclePlugin); ok { + if err := lc.Stop(); err != nil { + return fmt.Errorf("failed to stop plugin %s: %w", name, err) + } } } return nil diff --git a/pkg/plugins/registry_test.go b/pkg/plugins/registry_test.go new file mode 100644 index 0000000..23f07fd --- /dev/null +++ b/pkg/plugins/registry_test.go @@ -0,0 +1,48 @@ +package plugins + +import "testing" + +// coreOnly implements only the core Plugin interface: no lifecycle, no routes. +type coreOnly struct{ name string } + +func (c *coreOnly) Info() PluginInfo { return PluginInfo{Name: c.name} } + +// lifecyclePlugin additionally manages start/stop and records that it ran. +type lifecyclePlugin struct { + coreOnly + started, stopped bool +} + +func (l *lifecyclePlugin) Start() error { l.started = true; return nil } +func (l *lifecyclePlugin) Stop() error { l.stopped = true; return nil } + +func TestRegistrySkipsNonLifecyclePlugins(t *testing.T) { + r := NewRegistry("") + core := &coreOnly{name: "firewall-like"} + life := &lifecyclePlugin{coreOnly: coreOnly{name: "service-like"}} + + if err := r.Register(core); err != nil { + t.Fatal(err) + } + if err := r.Register(life); err != nil { + t.Fatal(err) + } + + // StartAll/StopAll must not require a non-lifecycle plugin to start or stop. + if err := r.StartAll(); err != nil { + t.Fatalf("StartAll() = %v", err) + } + if err := r.StopAll(); err != nil { + t.Fatalf("StopAll() = %v", err) + } + if !life.started || !life.stopped { + t.Error("the lifecycle plugin should have been started and stopped") + } + + if len(r.Plugins()) != 2 { + t.Errorf("Plugins() returned %d, want 2", len(r.Plugins())) + } + if len(r.List()) != 2 { + t.Errorf("List() returned %d, want 2", len(r.List())) + } +} diff --git a/pkg/plugins/types.go b/pkg/plugins/types.go index 03a4ed6..cfb75a8 100644 --- a/pkg/plugins/types.go +++ b/pkg/plugins/types.go @@ -80,16 +80,41 @@ type ResourceRequirements struct { RecommendedCPU string `json:"recommended_cpu" yaml:"recommended_cpu"` } +// Plugin is the core interface every plugin implements: its identity and metadata. +// Everything beyond identity is opt-in through the capability interfaces below, so a plugin +// implements only what it needs. A firewall, for example, has no services to start or stop, +// so it does not implement LifecyclePlugin. type Plugin interface { Info() PluginInfo +} + +// ConfigurablePlugin accepts runtime configuration at startup. +type ConfigurablePlugin interface { Initialize(config map[string]interface{}) error +} + +// LifecyclePlugin manages long-running state and is started and stopped with the agent. +type LifecyclePlugin interface { Start() error Stop() error - GetCapabilities() []Capability +} + +// RoutablePlugin serves its own HTTP endpoints; the host gives it the route group to mount on. +type RoutablePlugin interface { RegisterRoutes(router *gin.RouterGroup) error +} + +// WidgetPlugin contributes dashboard widget data for a deployment. +type WidgetPlugin interface { GetWidgetData(deploymentName string) (interface{}, error) } +// CapablePlugin declares the capabilities it provides. +type CapablePlugin interface { + GetCapabilities() []Capability +} + +// DeploymentPlugin creates and manages deployments from the plugin. type DeploymentPlugin interface { Plugin CreateDeployment(name string, config map[string]interface{}) (*DeploymentResult, error) diff --git a/pkg/pluginsdk/sdk.go b/pkg/pluginsdk/sdk.go new file mode 100644 index 0000000..fdc821e --- /dev/null +++ b/pkg/pluginsdk/sdk.go @@ -0,0 +1,67 @@ +// Package pluginsdk is imported by FlatRun plugin binaries to serve their routes over the +// host-provided socket. See examples/plugins/hello for usage. +package pluginsdk + +import ( + "context" + "encoding/json" + "fmt" + "net" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/flatrun/agent/pkg/pluginapi" +) + +// Serve runs the plugin until the host stops it. handler serves the plugin's own routes; +// pass nil if the plugin only reports info. It returns an error if the process was not +// launched by the host (the socket env var is unset) or the socket cannot be served. +func Serve(info pluginapi.Info, handler http.Handler) error { + socket := os.Getenv(pluginapi.EnvSocket) + if socket == "" { + return fmt.Errorf("not launched by the flatrun plugin host: %s is unset", pluginapi.EnvSocket) + } + handshake := os.Getenv(pluginapi.EnvHandshake) + + mux := http.NewServeMux() + mux.HandleFunc(pluginapi.InfoPath, func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set(pluginapi.HandshakeHeader, handshake) + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(info) + }) + if handler != nil { + mux.Handle("/", handler) + } + + // A stale socket file from a previous run would block Listen. + _ = os.Remove(socket) + ln, err := net.Listen("unix", socket) + if err != nil { + return fmt.Errorf("listen on plugin socket: %w", err) + } + + srv := &http.Server{Handler: mux} + + stop := make(chan os.Signal, 1) + signal.Notify(stop, syscall.SIGTERM, syscall.SIGINT) + go func() { + <-stop + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = srv.Shutdown(ctx) + }() + + if err := srv.Serve(ln); err != nil && err != http.ErrServerClosed { + return err + } + return nil +} + +// AgentCallback returns the agent API base URL and token a plugin can use to call back into +// the agent. Either may be empty when the host did not grant callback access. +func AgentCallback() (baseURL, token string) { + return os.Getenv(pluginapi.EnvAgentURL), os.Getenv(pluginapi.EnvToken) +} From 2f9445fd2c83e750002beef0abb59e059258865e Mon Sep 17 00:00:00 2001 From: nfebe Date: Tue, 30 Jun 2026 15:26:47 +0100 Subject: [PATCH 3/4] fix(nginx): Stop doubling the upstream for a deployment's own service An additional domain that points at a deployment's own primary service produced a proxy upstream with the deployment name repeated, which does not resolve, so the deployment became unreachable and requests timed out. Such a domain now routes to the correct container. --- internal/docker/compose_yaml.go | 5 +++++ internal/docker/compose_yaml_test.go | 2 ++ 2 files changed, 7 insertions(+) diff --git a/internal/docker/compose_yaml.go b/internal/docker/compose_yaml.go index 03d8e4d..3f318c2 100644 --- a/internal/docker/compose_yaml.go +++ b/internal/docker/compose_yaml.go @@ -167,6 +167,11 @@ func ContainerNameForService(content, deploymentName, service string) string { if deploymentName == "" || service == "" { return service } + // Already the primary container name (= deploymentName) or an already-scoped container + // name; return as-is so resolution is idempotent and never doubles the name. + if service == deploymentName || strings.HasPrefix(service, deploymentName+"-") { + return service + } scoped := fmt.Sprintf("%s-%s", deploymentName, service) compose, err := ParseComposeYAML(content) diff --git a/internal/docker/compose_yaml_test.go b/internal/docker/compose_yaml_test.go index f0f98eb..55ffe53 100644 --- a/internal/docker/compose_yaml_test.go +++ b/internal/docker/compose_yaml_test.go @@ -649,6 +649,8 @@ services: {"missing compose treats app as primary", "", "shop", "app", "shop"}, {"empty deployment returns bare service", multi, "", "app", "app"}, {"no explicit name uses primary rule", "name: x\nservices:\n app:\n image: nginx\n api:\n image: go\n", "x", "api", "x-api"}, + {"service already equals deployment name is not doubled", "", "shop", "shop", "shop"}, + {"already-scoped name is returned as-is", "", "shop", "shop-worker", "shop-worker"}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { From 364f5f44208f724113b577ee8c414a6893a7b400 Mon Sep 17 00:00:00 2001 From: nfebe Date: Tue, 30 Jun 2026 15:27:46 +0100 Subject: [PATCH 4/4] docs(agent): Trim the firewall package comment --- pkg/plugins/firewall/firewall.go | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/pkg/plugins/firewall/firewall.go b/pkg/plugins/firewall/firewall.go index b6a577e..c040b7d 100644 --- a/pkg/plugins/firewall/firewall.go +++ b/pkg/plugins/firewall/firewall.go @@ -1,11 +1,6 @@ -// Package firewall is a built-in FlatRun app (plugin) that models a host-wide firewall: -// default inbound/outbound stance plus allow/deny rules for traffic to and from the whole -// server. It operates at a different layer than the security module, which controls HTTP -// ingress at the proxy. -// -// This is a scaffold: rules are persisted, validated, and previewable, but Apply is a no-op. -// Enforcement will translate rules to nftables/iptables; until then nothing touches the host -// firewall, which avoids the obvious foot-gun of locking the operator out over SSH. +// Package firewall is a built-in app modelling a host-wide inbound/outbound firewall. +// Scaffold only: rules are persisted and validated, Apply is a no-op (enforcement via +// nftables/iptables is not wired yet). package firewall import (