diff --git a/config/config.go b/config/config.go index 16097fde..81245bf5 100644 --- a/config/config.go +++ b/config/config.go @@ -444,6 +444,45 @@ func Set(c *Configuration) { _config = c } +// ResolveToken populates the derived Token field, preferring values pinned +// through the environment over those in the configuration itself. +// +// Set remote when the values came from the Panel. Local values may use +// "file://" or "$VAR" indirection; expanding one sent over the network would +// leak files and environment variables back out through the token we attach to +// every request. Environment overrides must already match remote values so a +// configuration update cannot leave Wings and the Panel using different keys. +func (c *Configuration) ResolveToken(remote bool) error { + resolve := func(name, env, local string) (string, error) { + if remote && (strings.Contains(local, "$") || strings.HasPrefix(local, "file://")) { + return "", fmt.Errorf("config: remote %s cannot use token indirection", name) + } + if env != "" { + value, err := Expand(env) + if err != nil { + return "", err + } + if remote && value != local { + return "", fmt.Errorf("config: remote %s does not match environment override", name) + } + return value, nil + } + if remote { + return local, nil + } + return Expand(local) + } + + var err error + if c.Token.ID, err = resolve("token ID", os.Getenv("WINGS_TOKEN_ID"), c.AuthenticationTokenId); err != nil { + return err + } + if c.Token.Token, err = resolve("token", os.Getenv("WINGS_TOKEN"), c.AuthenticationToken); err != nil { + return err + } + return nil +} + // SetDebugViaFlag tracks if the application is running in debug mode because of // a command line flag argument. If so we do not want to store that configuration // change to the disk. @@ -614,23 +653,7 @@ func FromFile(path string) error { return err } - c.Token = Token{ - ID: os.Getenv("WINGS_TOKEN_ID"), - Token: os.Getenv("WINGS_TOKEN"), - } - if c.Token.ID == "" { - c.Token.ID = c.AuthenticationTokenId - } - if c.Token.Token == "" { - c.Token.Token = c.AuthenticationToken - } - - c.Token.ID, err = Expand(c.Token.ID) - if err != nil { - return err - } - c.Token.Token, err = Expand(c.Token.Token) - if err != nil { + if err := c.ResolveToken(false); err != nil { return err } @@ -638,7 +661,6 @@ func FromFile(path string) error { Set(c) return nil } - // ConfigureDirectories ensures that all the system directories exist on the // system. These directories are created so that only the owner can read the data, // and no other users. @@ -863,7 +885,7 @@ func Expand(v string) (string, error) { b, err := os.ReadFile(p) if err != nil { - return "", nil + return "", err } v = string(bytes.TrimRight(bytes.TrimRight(b, "\r"), "\n")) } diff --git a/config/config_docker.go b/config/config_docker.go index df10679b..038d28e1 100644 --- a/config/config_docker.go +++ b/config/config_docker.go @@ -78,6 +78,28 @@ type DockerConfiguration struct { Cpu int64 `default:"100" json:"cpu" yaml:"cpu"` } `json:"installer_limits" yaml:"installer_limits"` + // CpuPeriod is the length of a CFS scheduling window in microseconds. Server + // quotas scale with it, so the configured CPU limits stay the same. A shorter + // period reduces the worst case throttle latency at the cost of additional + // scheduler overhead. + CpuPeriod int64 `default:"100000" json:"cpu_period" yaml:"cpu_period"` + + // CpuBurst allows containers to bank unused CFS quota within a period and spend + // it on short spikes without raising their long term CPU limit. Percent sizes the + // burst relative to a server's quota and is capped at 100 by the kernel. Requires + // Linux 5.14 or newer, it is skipped silently otherwise. + CpuBurst struct { + Enabled bool `default:"true" json:"enabled" yaml:"enabled"` + Percent int64 `default:"100" json:"percent" yaml:"percent"` + } `json:"cpu_burst" yaml:"cpu_burst"` + + // CpuShares is the relative CFS weight of server containers when the host is + // fully saturated, it limits nothing on an idle host. Zero leaves containers + // at the engine default. Wings historically set 1024, which cgroup v2 converts + // to less than half of the default weight, set that value to restore the old + // bias towards host system services. + CpuShares int64 `default:"0" json:"cpu_shares" yaml:"cpu_shares"` + // Overhead controls the memory overhead given to all containers to circumvent certain // software such as the JVM not staying below the maximum memory limit. Overhead Overhead `json:"overhead" yaml:"overhead"` @@ -101,6 +123,12 @@ type DockerConfiguration struct { } `json:"log_config" yaml:"log_config"` } +// CpuPeriodMicroseconds returns the configured CFS period clamped to the range +// the kernel accepts. +func (c DockerConfiguration) CpuPeriodMicroseconds() int64 { + return min(max(c.CpuPeriod, 1_000), 1_000_000) +} + func (c DockerConfiguration) ContainerLogConfig() container.LogConfig { if c.LogConfig.Type == "" { return container.LogConfig{} diff --git a/config/config_docker_test.go b/config/config_docker_test.go index 416ba0f7..de43cfe0 100644 --- a/config/config_docker_test.go +++ b/config/config_docker_test.go @@ -82,6 +82,28 @@ func TestDockerRegistryCredentialsForImage(t *testing.T) { } } +func TestCpuPeriodMicroseconds(t *testing.T) { + tests := []struct { + name string + period int64 + expected int64 + }{ + {name: "default period", period: 100_000, expected: 100_000}, + {name: "shorter period", period: 20_000, expected: 20_000}, + {name: "below kernel minimum", period: 500, expected: 1_000}, + {name: "above kernel maximum", period: 5_000_000, expected: 1_000_000}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := DockerConfiguration{CpuPeriod: tt.period} + if v := cfg.CpuPeriodMicroseconds(); v != tt.expected { + t.Errorf("expected %d, got %d", tt.expected, v) + } + }) + } +} + func TestDockerRegistryPathCredentialsDoNotMatchSiblingPath(t *testing.T) { cfg := DockerConfiguration{ Registries: map[string]RegistryConfiguration{ diff --git a/config/config_token_test.go b/config/config_token_test.go new file mode 100644 index 00000000..ad04a7f5 --- /dev/null +++ b/config/config_token_test.go @@ -0,0 +1,60 @@ +package config + +import ( + "os" + "path/filepath" + "testing" +) + +func TestResolveRemoteToken(t *testing.T) { + t.Setenv("WINGS_TOKEN_ID", "") + t.Setenv("WINGS_TOKEN", "") + + cfg := Configuration{ + AuthenticationTokenId: "panel-id", + AuthenticationToken: "panel-token", + } + if err := cfg.ResolveToken(true); err != nil { + t.Fatalf("expected remote credentials to resolve: %v", err) + } + if cfg.Token.ID != "panel-id" || cfg.Token.Token != "panel-token" { + t.Fatalf("unexpected resolved credentials: %#v", cfg.Token) + } +} + +func TestResolveRemoteTokenRejectsIndirection(t *testing.T) { + t.Setenv("WINGS_TOKEN_ID", "") + t.Setenv("WINGS_TOKEN", "") + + tests := []Configuration{ + {AuthenticationTokenId: "file:///tmp/id", AuthenticationToken: "panel-token"}, + {AuthenticationTokenId: "panel-id", AuthenticationToken: "$PANEL_TOKEN"}, + } + for _, cfg := range tests { + if err := cfg.ResolveToken(true); err == nil { + t.Fatal("expected remote token indirection to be rejected") + } + } +} + +func TestResolveRemoteTokenRequiresEnvironmentMatch(t *testing.T) { + secret := filepath.Join(t.TempDir(), "token") + if err := os.WriteFile(secret, []byte("panel-token\n"), 0o600); err != nil { + t.Fatal(err) + } + t.Setenv("WINGS_TOKEN_ID", "panel-id") + t.Setenv("WINGS_TOKEN", "file://"+secret) + + cfg := Configuration{ + AuthenticationTokenId: "panel-id", + AuthenticationToken: "panel-token", + } + if err := cfg.ResolveToken(true); err != nil { + t.Fatalf("expected matching environment credentials to resolve: %v", err) + } + + cfg.AuthenticationToken = "rotated-token" + if err := cfg.ResolveToken(true); err == nil { + t.Fatal("expected mismatched environment credentials to be rejected") + } +} \ No newline at end of file diff --git a/environment/docker/cgroup_burst.go b/environment/docker/cgroup_burst.go new file mode 100644 index 00000000..d8f05800 --- /dev/null +++ b/environment/docker/cgroup_burst.go @@ -0,0 +1,131 @@ +package docker + +import ( + "context" + "os" + "path" + "strconv" + "strings" + "sync" + + "emperror.dev/errors" + "github.com/apex/log" + "github.com/docker/docker/client" + + "github.com/pelican/wings/config" +) + +// cgroupV2 reports whether the host uses the unified cgroup v2 hierarchy. +var cgroupV2 = sync.OnceValue(func() bool { + _, err := os.Stat("/sys/fs/cgroup/cgroup.controllers") + return err == nil +}) + +var burstWarning sync.Once + +// cpuBurstMicroseconds returns the burst allowance in microseconds for the given +// CFS quota and configured percentage. The kernel rejects a burst larger than the +// quota, so the value is clamped to it. +func cpuBurstMicroseconds(quota int64, percent int64) int64 { + if quota <= 0 || percent <= 0 { + return 0 + } + if percent > 100 { + percent = 100 + } + return quota * percent / 100 +} + +// resolveCgroupCpuFile parses the contents of a /proc//cgroup file and +// returns the absolute path of the CFS burst file for that process's cgroup. +func resolveCgroupCpuFile(procCgroup string, v2 bool) (string, error) { + for _, line := range strings.Split(procCgroup, "\n") { + parts := strings.SplitN(line, ":", 3) + if len(parts) != 3 || !strings.HasPrefix(parts[2], "/") || strings.Contains(parts[2], "..") { + continue + } + if v2 { + if parts[0] == "0" && parts[1] == "" { + return path.Join("/sys/fs/cgroup", parts[2], "cpu.max.burst"), nil + } + continue + } + for _, controller := range strings.Split(parts[1], ",") { + if controller == "cpu" { + return path.Join("/sys/fs/cgroup/cpu", parts[2], "cpu.cfs_burst_us"), nil + } + } + } + return "", errors.New("environment/docker: no cpu controller found in cgroup file") +} + +// writeCpuBurst writes a burst value in microseconds into the cpu cgroup of the +// given process. This is expected to fail on kernels older than 5.14 or when the +// cgroup hierarchy is not writable by Wings, so failures are only logged. +func writeCpuBurst(l *log.Entry, pid int, burst int64) { + if pid <= 0 { + return + } + if err := writeBurstFile(pid, burst); err != nil { + logBurstFailure(l.WithField("error", err), burst) + return + } + l.WithField("burst_us", burst).Debug("updated container cpu burst") +} + +func writeBurstFile(pid int, burst int64) error { + b, err := os.ReadFile("/proc/" + strconv.Itoa(pid) + "/cgroup") + if err != nil { + return err + } + f, err := resolveCgroupCpuFile(string(b), cgroupV2()) + if err != nil { + return err + } + return os.WriteFile(f, []byte(strconv.FormatInt(burst, 10)), 0o644) +} + +// logBurstFailure warns the first time a burst cannot be applied and stays at +// debug otherwise. Failed clears are always quiet since a host that never +// accepted a burst has nothing to clear. +func logBurstFailure(l *log.Entry, burst int64) { + if burst > 0 { + first := false + burstWarning.Do(func() { first = true }) + if first { + l.Warn("failed to set cpu burst, this requires Linux 5.14 or newer and a writable cgroup hierarchy") + return + } + } + l.Debug("failed to set cpu burst") +} + +// SetCpuBurst applies the configured CFS burst to a running container based on +// the CFS quota in microseconds it was created with. This is a no-op when +// bursting is disabled or the container has no CPU limit. +func SetCpuBurst(ctx context.Context, cli *client.Client, containerID string, quota int64) { + cfg := config.Get().Docker.CpuBurst + if !cfg.Enabled || quota <= 0 { + return + } + c, err := cli.ContainerInspect(ctx, containerID) + if err != nil || c.State == nil { + return + } + writeCpuBurst(log.WithField("container_id", containerID), c.State.Pid, cpuBurstMicroseconds(quota, cfg.Percent)) +} + +// applyCpuBurst applies the configured CFS burst to the environment's container +// using its current CPU limit. +func (e *Environment) applyCpuBurst(ctx context.Context) { + quota := e.Configuration.Limits().CpuLimit * config.Get().Docker.CpuPeriodMicroseconds() / 100 + SetCpuBurst(ctx, e.client, e.Id, quota) +} + +// clearCpuBurst zeroes the CFS burst for the given container process. This must +// happen before a quota change is applied since the kernel rejects a quota lower +// than the current burst. It runs even when bursting is disabled so a value set +// before the feature was turned off cannot block future quota changes. +func (e *Environment) clearCpuBurst(pid int) { + writeCpuBurst(e.log(), pid, 0) +} \ No newline at end of file diff --git a/environment/docker/cgroup_burst_test.go b/environment/docker/cgroup_burst_test.go new file mode 100644 index 00000000..70cafd6e --- /dev/null +++ b/environment/docker/cgroup_burst_test.go @@ -0,0 +1,105 @@ +package docker + +import "testing" + +func TestCpuBurstMicroseconds(t *testing.T) { + tests := []struct { + name string + quota int64 + percent int64 + expected int64 + }{ + {name: "full quota", quota: 200_000, percent: 100, expected: 200_000}, + {name: "half quota", quota: 200_000, percent: 50, expected: 100_000}, + {name: "zero percent", quota: 200_000, percent: 0, expected: 0}, + {name: "percent above kernel cap", quota: 200_000, percent: 150, expected: 200_000}, + {name: "negative percent", quota: 200_000, percent: -50, expected: 0}, + {name: "no quota", quota: 0, percent: 100, expected: 0}, + {name: "negative quota", quota: -1, percent: 100, expected: 0}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if v := cpuBurstMicroseconds(tt.quota, tt.percent); v != tt.expected { + t.Errorf("expected %d, got %d", tt.expected, v) + } + }) + } +} + +func TestResolveCgroupCpuFile(t *testing.T) { + tests := []struct { + name string + procCgroup string + v2 bool + expected string + wantErr bool + }{ + { + name: "v2 systemd scope", + procCgroup: "0::/system.slice/docker-abc123.scope\n", + v2: true, + expected: "/sys/fs/cgroup/system.slice/docker-abc123.scope/cpu.max.burst", + }, + { + name: "v2 rootless", + procCgroup: "0::/user.slice/user-1000.slice/user@1000.service/user.slice/docker-abc123.scope\n", + v2: true, + expected: "/sys/fs/cgroup/user.slice/user-1000.slice/user@1000.service/user.slice/docker-abc123.scope/cpu.max.burst", + }, + { + name: "v1 combined cpu controller", + procCgroup: "12:pids:/docker/abc123\n4:cpu,cpuacct:/docker/abc123\n1:name=systemd:/docker/abc123\n", + v2: false, + expected: "/sys/fs/cgroup/cpu/docker/abc123/cpu.cfs_burst_us", + }, + { + name: "v1 bare cpu controller", + procCgroup: "4:cpu:/docker/abc123\n", + v2: false, + expected: "/sys/fs/cgroup/cpu/docker/abc123/cpu.cfs_burst_us", + }, + { + name: "v1 cpuset and cpuacct do not match", + procCgroup: "5:cpuset:/docker/abc123\n4:cpuacct:/docker/abc123\n", + v2: false, + wantErr: true, + }, + { + name: "v1 host ignores the unified hierarchy line", + procCgroup: "0::/docker/abc123\n", + v2: false, + wantErr: true, + }, + { + name: "namespaced relative path", + procCgroup: "0::/../../system.slice\n", + v2: true, + wantErr: true, + }, + { + name: "empty content", + procCgroup: "", + v2: true, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + f, err := resolveCgroupCpuFile(tt.procCgroup, tt.v2) + if tt.wantErr { + if err == nil { + t.Errorf("expected an error, got %q", f) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %s", err) + } + if f != tt.expected { + t.Errorf("expected %q, got %q", tt.expected, f) + } + }) + } +} \ No newline at end of file diff --git a/environment/docker/container.go b/environment/docker/container.go index 541923ac..49b09bfb 100644 --- a/environment/docker/container.go +++ b/environment/docker/container.go @@ -108,7 +108,8 @@ func (e *Environment) InSituUpdate() error { ctx, cancel := context.WithTimeout(context.Background(), time.Second*10) defer cancel() - if _, err := e.ContainerInspect(ctx); err != nil { + c, err := e.ContainerInspect(ctx) + if err != nil { // If the container doesn't exist for some reason there really isn't anything // we can do to fix that in this process (it doesn't make sense at least). In those // cases just return without doing anything since we still want to save the configuration @@ -121,6 +122,12 @@ func (e *Environment) InSituUpdate() error { return errors.Wrap(err, "environment/docker: could not inspect container") } + // The kernel rejects a CFS quota lower than the current burst, so remove the + // burst before updating the limits and re-apply it afterwards. + if c.State != nil { + e.clearCpuBurst(c.State.Pid) + } + // CPU pinning cannot be removed once it is applied to a container. The same is true // for removing memory limits, a container must be re-created. // @@ -130,6 +137,8 @@ func (e *Environment) InSituUpdate() error { }); err != nil { return errors.Wrap(err, "environment/docker: could not update container") } + + e.applyCpuBurst(ctx) return nil } diff --git a/environment/docker/power.go b/environment/docker/power.go index 72c26fa2..301d8786 100644 --- a/environment/docker/power.go +++ b/environment/docker/power.go @@ -75,6 +75,7 @@ func (e *Environment) Start(ctx context.Context) error { // If the server is running update our internal state and continue on with the attach. if c.State.Running { e.SetState(environment.ProcessRunningState) + e.applyCpuBurst(ctx) return e.Attach(ctx) } @@ -121,7 +122,8 @@ func (e *Environment) Start(ctx context.Context) error { if err := e.client.ContainerStart(actx, e.Id, container.StartOptions{}); err != nil { return errors.WrapIf(err, "environment/docker: failed to start container") } - + e.applyCpuBurst(actx) + // No errors, good to continue through. sawError = false return nil diff --git a/environment/settings.go b/environment/settings.go index d090ae77..8ce918c1 100644 --- a/environment/settings.go +++ b/environment/settings.go @@ -128,9 +128,11 @@ func (l Limits) AsContainerResources() container.Resources { // // @see https://github.com/pterodactyl/panel/issues/3988 if l.CpuLimit > 0 { - resources.CPUQuota = l.CpuLimit * 1_000 - resources.CPUPeriod = 100_000 - resources.CPUShares = 1024 + cfg := config.Get().Docker + period := cfg.CpuPeriodMicroseconds() + resources.CPUQuota = l.CpuLimit * period / 100 + resources.CPUPeriod = period + resources.CPUShares = cfg.CpuShares } // Similar to above, don't set the specific assigned CPUs if we didn't actually limit diff --git a/remote/http.go b/remote/http.go index 5893ebc5..90b778f4 100644 --- a/remote/http.go +++ b/remote/http.go @@ -8,6 +8,7 @@ import ( "net/http" "strconv" "strings" + "sync" "time" "github.com/pelican/wings/internal/models" @@ -34,11 +35,13 @@ type Client interface { ValidateSftpCredentials(ctx context.Context, request SftpAuthRequest) (SftpAuthResponse, error) SendActivityLogs(ctx context.Context, activity []models.Activity) error PushServerStateChange(ctx context.Context, sid string, stateChange ServerStateChange) error + SetCredentials(id, token string) } type client struct { httpClient *http.Client baseUrl string + mu sync.RWMutex tokenId string token string maxAttempts int @@ -70,6 +73,22 @@ func WithCredentials(id, token string) ClientOption { } } +// SetCredentials replaces the credentials used when making requests to the +// remote API endpoint. +func (c *client) SetCredentials(id, token string) { + c.mu.Lock() + defer c.mu.Unlock() + c.tokenId = id + c.token = token +} + +// credentials returns the credentials currently in use by this client. +func (c *client) credentials() (string, string) { + c.mu.RLock() + defer c.mu.RUnlock() + return c.tokenId, c.token +} + // WithCustomHeaders sets custom headers to be used when making remote requests. func WithCustomHeaders(headers map[string]string) ClientOption { return func(c *client) { @@ -114,10 +133,11 @@ func (c *client) requestOnce(ctx context.Context, method, path string, body io.R return nil, err } - req.Header.Set("User-Agent", fmt.Sprintf("Pelican Wings/v%s (id:%s)", system.Version, c.tokenId)) + tokenId, token := c.credentials() + req.Header.Set("User-Agent", fmt.Sprintf("Pelican Wings/v%s (id:%s)", system.Version, tokenId)) req.Header.Set("Accept", "application/json") req.Header.Set("Content-Type", "application/json") - req.Header.Set("Authorization", fmt.Sprintf("Bearer %s.%s", c.tokenId, c.token)) + req.Header.Set("Authorization", fmt.Sprintf("Bearer %s.%s", tokenId, token)) // Apply custom headers, but prevent overriding critical headers criticalHeaders := map[string]bool{ diff --git a/remote/http_test.go b/remote/http_test.go index 8003b943..81cdcc3b 100644 --- a/remote/http_test.go +++ b/remote/http_test.go @@ -35,6 +35,28 @@ func TestRequest(t *testing.T) { assert.NotNil(t, r) } +func TestSetCredentials(t *testing.T) { + var authorization []string + c, server := createTestClient(func(rw http.ResponseWriter, r *http.Request) { + authorization = append(authorization, r.Header.Get("Authorization")) + rw.WriteHeader(http.StatusOK) + }) + defer server.Close() + + if _, err := c.requestOnce(context.Background(), http.MethodGet, "/test", nil); err != nil { + t.Fatal(err) + } + c.SetCredentials("rotated-id", "rotated-token") + if _, err := c.requestOnce(context.Background(), http.MethodGet, "/test", nil); err != nil { + t.Fatal(err) + } + + assert.Equal(t, []string{ + "Bearer testid.testtoken", + "Bearer rotated-id.rotated-token", + }, authorization) +} + func TestRequestRetry(t *testing.T) { // Test if the client attempts failed requests i := 0 diff --git a/router/router_server_backup_test.go b/router/router_server_backup_test.go index 68dd9e84..e1eac41e 100644 --- a/router/router_server_backup_test.go +++ b/router/router_server_backup_test.go @@ -27,6 +27,7 @@ func init() { type backupTestRemoteClient struct { restoreStatus chan string + credentials chan [2]string } func (c backupTestRemoteClient) GetBackupRemoteUploadURLs(context.Context, string, int64) (remote.BackupRemoteUploadResponse, error) { @@ -83,11 +84,16 @@ func (c backupTestRemoteClient) SendActivityLogs(context.Context, []models.Activ return nil } - func (c backupTestRemoteClient) PushServerStateChange(context.Context, string, remote.ServerStateChange) error { return nil } +func (c backupTestRemoteClient) SetCredentials(id, token string) { + if c.credentials != nil { + c.credentials <- [2]string{id, token} + } +} + type backupTestEnvironment struct{} func (backupTestEnvironment) Type() string { return "test" } diff --git a/router/router_system.go b/router/router_system.go index c27eb289..ed99c6bd 100644 --- a/router/router_system.go +++ b/router/router_system.go @@ -244,6 +244,22 @@ func postUpdateConfiguration(c *gin.Context) { cfg.Api.Ssl.CertificateFile = config.Get().Api.Ssl.CertificateFile } + // The token that everything authenticates against is a derived value that is + // not part of the payload sent by the Panel, so it has to be re-resolved from + // the new token values. + if err := cfg.ResolveToken(true); err != nil { + middleware.CaptureAndAbort(c, err) + return + } + + // Refuse to go any further with a token we could never authenticate against. + if cfg.Token.ID == "" || cfg.Token.Token == "" { + middleware.CaptureAndAbort(c, errors.New("config: refusing to apply an update with an empty authentication token")) + return + } + + tokenId, token := cfg.Token.ID, cfg.Token.Token + // Try to write this new configuration to the disk before updating our global // state with it. if err := config.WriteToDisk(cfg); err != nil { @@ -253,6 +269,11 @@ func postUpdateConfiguration(c *gin.Context) { // Since we wrote it to the disk successfully now update the global configuration // state to use this new configuration struct. config.Set(cfg) + + // Requests we make back to the Panel use credentials that were captured when + // the client was created at boot, so they have to be rotated explicitly. + middleware.ExtractManager(c).Client().SetCredentials(tokenId, token) + c.JSON(http.StatusOK, postUpdateConfigurationResponse{ Applied: true, }) diff --git a/router/router_system_test.go b/router/router_system_test.go new file mode 100644 index 00000000..453bd23c --- /dev/null +++ b/router/router_system_test.go @@ -0,0 +1,55 @@ +package router + +import ( + "net/http/httptest" + "path/filepath" + "strings" + "testing" + + "github.com/gin-gonic/gin" + + "github.com/pelican/wings/config" + "github.com/pelican/wings/server" +) + +func TestPostUpdateConfigurationRotatesCredentials(t *testing.T) { + t.Setenv("WINGS_TOKEN_ID", "") + t.Setenv("WINGS_TOKEN", "") + + cfg, err := config.NewAtPath(filepath.Join(t.TempDir(), "config.yml")) + if err != nil { + t.Fatal(err) + } + cfg.AuthenticationTokenId = "old-id" + cfg.AuthenticationToken = "old-token" + if err := cfg.ResolveToken(false); err != nil { + t.Fatal(err) + } + config.Set(cfg) + + credentials := make(chan [2]string, 1) + manager := server.NewEmptyManager(backupTestRemoteClient{credentials: credentials}) + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Set("manager", manager) + c.Request = httptest.NewRequest("POST", "/api/update", strings.NewReader(`{"token_id":"new-id","token":"new-token"}`)) + c.Request.Header.Set("Content-Type", "application/json") + + postUpdateConfiguration(c) + + if recorder.Code != 200 { + t.Fatalf("expected successful update, got status %d", recorder.Code) + } + updated := config.Get() + if updated.Token.ID != "new-id" || updated.Token.Token != "new-token" { + t.Fatalf("unexpected resolved credentials: %#v", updated.Token) + } + select { + case got := <-credentials: + if got != [2]string{"new-id", "new-token"} { + t.Fatalf("unexpected client credentials: %#v", got) + } + default: + t.Fatal("expected client credentials to be rotated") + } +} \ No newline at end of file diff --git a/server/install.go b/server/install.go index e1237b8e..c86abb51 100644 --- a/server/install.go +++ b/server/install.go @@ -24,6 +24,7 @@ import ( "github.com/pelican/wings/config" "github.com/pelican/wings/environment" + "github.com/pelican/wings/environment/docker" "github.com/pelican/wings/remote" "github.com/pelican/wings/system" ) @@ -516,6 +517,8 @@ func (ip *InstallationProcess) Execute() (string, error) { return "", err } + docker.SetCpuBurst(ctx, ip.client, r.ID, hostConf.Resources.CPUQuota) + // Process the install event in the background by listening to the stream output until the // container has stopped, at which point we'll disconnect from it. // diff --git a/sftp/event.go b/sftp/event.go index ecfedcbc..60c6c797 100644 --- a/sftp/event.go +++ b/sftp/event.go @@ -32,9 +32,8 @@ func (eh *eventHandler) Log(e models.Event, fa FileAction) error { "files": []string{fa.Entity}, } if fa.Target != "" { - metadata = map[string]interface{}{ - "from": fa.Entity, - "to": fa.Target, + metadata["files"] = []map[string]string{ + {"from": fa.Entity, "to": fa.Target}, } } @@ -57,4 +56,4 @@ func (eh *eventHandler) MustLog(e models.Event, fa FileAction) { if err := eh.Log(e, fa); err != nil { log.WithField("error", errors.WithStack(err)).WithField("event", e).Error("sftp: failed to log event") } -} +} \ No newline at end of file diff --git a/sftp/handler.go b/sftp/handler.go index 4a1ceaf9..db665473 100644 --- a/sftp/handler.go +++ b/sftp/handler.go @@ -23,6 +23,7 @@ const ( PermissionFileCreate = "file.create" PermissionFileUpdate = "file.update" PermissionFileDelete = "file.delete" + sftpAttributeExtended = 1 << 31 ) type Handler struct { @@ -105,7 +106,7 @@ func (h *Handler) Fileread(request *sftp.Request) (io.ReaderAt, error) { defer h.mu.Unlock() if err := h.fs.IsIgnored(request.Filepath); err != nil { return nil, err - } + } f, _, err := h.fs.File(request.Filepath) if err != nil { if !errors.Is(err, os.ErrNotExist) { @@ -134,7 +135,7 @@ func (h *Handler) Filewrite(request *sftp.Request) (io.WriterAt, error) { if err := h.fs.IsIgnored(request.Filepath); err != nil { return nil, err - } + } // The specific permission required to perform this action. If the file exists on the // system already it only needs to be an update, otherwise we'll check for a create. permission := PermissionFileUpdate @@ -168,6 +169,29 @@ func (h *Handler) Filewrite(request *sftp.Request) (io.WriterAt, error) { return quotaWriterAt{WriterAt: f, server: h.server}, nil } +func setstatMode(request *sftp.Request) (os.FileMode, error) { + // pkg/sftp allocates the client-provided extended attribute count before + // validating the remaining packet length. Reject it before parsing to avoid + // allowing a small packet to request an effectively unbounded allocation. + if request.Flags&sftpAttributeExtended != 0 { + return 0, sftp.ErrSSHFxBadMessage + } + attrs := request.Attributes() + if attrs == nil { + return 0, sftp.ErrSSHFxBadMessage + } + mode := attrs.FileMode().Perm() + // If the client passes an invalid FileMode just use the default 0644. + if mode == 0o000 { + mode = os.FileMode(0o644) + } + // Force directories to be 0755. + if attrs.FileMode().IsDir() { + mode = 0o755 + } + return mode, nil +} + // Filecmd hander for basic SFTP system calls related to files, but not anything to do with reading // or writing to those files. func (h *Handler) Filecmd(request *sftp.Request) error { @@ -178,10 +202,6 @@ func (h *Handler) Filecmd(request *sftp.Request) error { if request.Target != "" { l = l.WithField("target", request.Target) } - - if err := h.fs.IsIgnored(request.Filepath); err != nil { - return err - } switch request.Method { // Allows a user to make changes to the permissions of a given file or directory @@ -190,14 +210,9 @@ func (h *Handler) Filecmd(request *sftp.Request) error { if !h.can(PermissionFileUpdate) { return sftp.ErrSSHFxPermissionDenied } - mode := request.Attributes().FileMode().Perm() - // If the client passes an invalid FileMode just use the default 0644. - if mode == 0o000 { - mode = os.FileMode(0o644) - } - // Force directories to be 0755. - if request.Attributes().FileMode().IsDir() { - mode = 0o755 + mode, err := setstatMode(request) + if err != nil { + return err } if err := h.fs.Chmod(request.Filepath, mode); err != nil { if errors.Is(err, os.ErrNotExist) { diff --git a/sftp/handler_test.go b/sftp/handler_test.go index 08c8bf49..e3a39d7d 100644 --- a/sftp/handler_test.go +++ b/sftp/handler_test.go @@ -1,10 +1,12 @@ package sftp import ( + "encoding/binary" "errors" "io" "testing" + "github.com/apex/log" pkgsftp "github.com/pkg/sftp" "github.com/pelican/wings/server" @@ -137,4 +139,58 @@ func TestWriterForwardsWritesWhenServerIsAvailable(t *testing.T) { if n != 4 { t.Fatalf("expected forwarded byte count, got %d", n) } +} + +func TestHandlerRejectsMalformedSetstatAttributes(t *testing.T) { + srv, err := server.New(nil) + if err != nil { + t.Fatal(err) + } + h := Handler{ + server: srv, + permissions: []string{PermissionFileUpdate}, + logger: log.WithField("test", t.Name()), + } + request := pkgsftp.NewRequest("Setstat", "/") + request.Flags = 1 // SSH_FILEXFER_ATTR_SIZE + + if err := h.Filecmd(request); !errors.Is(err, pkgsftp.ErrSSHFxBadMessage) { + t.Fatalf("expected bad message, got %v", err) + } + + request.Flags = sftpAttributeExtended + request.Attrs = make([]byte, 4) + binary.BigEndian.PutUint32(request.Attrs, ^uint32(0)) + if err := h.Filecmd(request); !errors.Is(err, pkgsftp.ErrSSHFxBadMessage) { + t.Fatalf("expected extended attributes to be rejected, got %v", err) + } +} + +func TestSetstatMode(t *testing.T) { + tests := []struct { + name string + mode uint32 + expected uint32 + }{ + {name: "file permissions", mode: 0o600, expected: 0o600}, + {name: "default permissions", mode: 0o000, expected: 0o644}, + {name: "directory permissions", mode: 0o040700, expected: 0o755}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + request := pkgsftp.NewRequest("Setstat", "/test") + request.Flags = 4 // SSH_FILEXFER_ATTR_PERMISSIONS + request.Attrs = make([]byte, 4) + binary.BigEndian.PutUint32(request.Attrs, tt.mode) + + mode, err := setstatMode(request) + if err != nil { + t.Fatal(err) + } + if uint32(mode) != tt.expected { + t.Fatalf("expected mode %04o, got %04o", tt.expected, mode) + } + }) + } } \ No newline at end of file diff --git a/sftp/server.go b/sftp/server.go index d40cf3c4..8978cf2e 100644 --- a/sftp/server.go +++ b/sftp/server.go @@ -126,7 +126,7 @@ func (c *SFTPServer) AcceptInbound(conn net.Conn, config *ssh.ServerConfig) erro go ssh.DiscardRequests(reqs) for ch := range chans { - // If not a session channel we just move on because it's not something we + // If its not a session channel we just move on because its not something we // know how to handle at this point. if ch.ChannelType() != "session" { _ = ch.Reject(ssh.UnknownChannelType, "unknown channel type") @@ -154,6 +154,7 @@ func (c *SFTPServer) AcceptInbound(conn net.Conn, config *ssh.ServerConfig) erro } } } + return nil } @@ -231,7 +232,7 @@ func (c *SFTPServer) makeCredentialsRequest(conn ssh.ConnMetadata, t remote.Sftp logger.Warn("failed to validate user credentials (password authentication is disabled; only SSH keys are allowed)") return nil, &remote.SftpKeyOnlyError{} } - + resp, err := c.manager.Client().ValidateSftpCredentials(context.Background(), request) if err != nil { if _, ok := err.(*remote.SftpInvalidCredentialsError); ok { @@ -258,4 +259,4 @@ func (c *SFTPServer) makeCredentialsRequest(conn ssh.ConnMetadata, t remote.Sftp // PrivateKeyPath returns the path the host private key for this server instance. func (c *SFTPServer) PrivateKeyPath() string { return path.Join(c.BasePath, ".sftp/id_ed25519") -} +} \ No newline at end of file diff --git a/sftp/utils.go b/sftp/utils.go index 88295016..f78252a5 100644 --- a/sftp/utils.go +++ b/sftp/utils.go @@ -3,6 +3,7 @@ package sftp import ( "io" "os" + "reflect" ) const ( @@ -30,6 +31,23 @@ func (l ListerAt) ListAt(f []os.FileInfo, offset int64) (int, error) { type fxErr uint32 +func (e fxErr) As(target interface{}) bool { + // pkg/sftp checks errors against its private fxerr type before writing status packets. + v := reflect.ValueOf(target) + if v.Kind() != reflect.Ptr || v.IsNil() { + return false + } + + elem := v.Elem() + t := elem.Type() + if elem.Kind() != reflect.Uint32 || t.PkgPath() != "github.com/pkg/sftp" || t.Name() != "fxerr" { + return false + } + + elem.SetUint(uint64(e)) + return true +} + func (e fxErr) Error() string { switch e { case ErrSSHQuotaExceeded: @@ -37,4 +55,4 @@ func (e fxErr) Error() string { default: return "Failure" } -} +} \ No newline at end of file