Finalize activity event sending logic and cron config

* fix: arm64 docker builds

Don't hardcode amd64 platform for the Wings binary.

* update docker file

don't specify buildplatform
remove upx as it causes arm64 failures
remove goos as the build is on linux hosts.

Co-authored-by: softwarenoob <admin@softwarenoob.com>

CHANGELOG.md

@@ -1,5 +1,37 @@
 # Changelog
 
+## v1.6.4
+### Fixed
+* Fixes a bug causing CPU limiting to not be properly applied to servers.
+* Fixes a bug causing zip archives to decompress without taking into account nested folder structures.
+
+## v1.6.3
+### Fixed
+* Fixes SFTP authentication failing for administrative users due to a permissions adjustment on the Panel.
+
+## v1.6.2
+### Fixed
+* Fixes file upload size not being properly enforced.
+* Fixes a bug that prevented listing a directory when it contained a named pipe. Also added a check to prevent attempting to read a named pipe directly.
+* Fixes a bug with the archiver logic that would include folders that had the same name prefix. (for example, requesting only `map` would also include `map2` and `map3`)
+* Requests to the Panel that return a client error (4xx response code) no longer trigger an exponential backoff, they immediately stop the request.
+
+### Changed
+* CPU limit fields are only set on the Docker container if they have been specified for the server — otherwise they are left empty.
+
+### Added
+* Added the ability to define the location of the temporary folder used by Wings — defaults to `/tmp/pterodactyl`.
+* Adds the ability to authenticate for SFTP using public keys (requires `Panel@1.8.0`).
+
+## v1.6.1
+### Fixed
+* Fixes error that would sometimes occur when starting a server that would cause the temporary power action lock to never be released due to a blocked channel.
+* Fixes a bug causing the CPU usage of Wings to get stuck at 100% when a server is deleted while the installation process is running.
+
+### Changed
+* Cleans up a lot of the logic for handling events between the server and environment process to make it easier to make modifications to down the road.
+* Cleans up logic handling the `StopAndWait` logic for stopping a server gracefully before terminating the process if it does not respond.
+
 ## v1.6.0
 ### Fixed
 * Internal logic for processing a server start event has been adjusted to attach to the Docker container before attempting to start the container. This should fix issues where a server would get stuck after pulling the container image.

Dockerfile

@@ -1,19 +1,18 @@
 # Stage 1 (Build)
-FROM --platform=$BUILDPLATFORM golang:1.17-alpine AS builder
+FROM golang:1.17-alpine AS builder
 
 ARG VERSION
-RUN apk add --update --no-cache git make upx
+RUN apk add --update --no-cache git make
 WORKDIR /app/
 COPY go.mod go.sum /app/
 RUN go mod download
 COPY . /app/
-RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build \
+RUN CGO_ENABLED=0 go build \
     -ldflags="-s -w -X github.com/pterodactyl/wings/system.Version=$VERSION" \
     -v \
     -trimpath \
     -o wings \
     wings.go
-RUN upx wings
 RUN echo "ID=\"distroless\"" > /etc/os-release
 
 # Stage 2 (Final)

Makefile

@@ -6,7 +6,7 @@ build:
 
 debug:
 	go build -ldflags="-X github.com/pterodactyl/wings/system.Version=$(GIT_HEAD)"
-	sudo ./wings --debug --ignore-certificate-errors --config config.yml --pprof
+	sudo ./wings --debug --ignore-certificate-errors --config config.yml --pprof --pprof-block-rate 1
 
 # Runs a remotly debuggable session for Wings allowing an IDE to connect and target
 # different breakpoints.

README.md

@@ -19,22 +19,16 @@ I would like to extend my sincere thanks to the following sponsors for helping f
 | Company | About |
 | ------- | ----- |
 | [**WISP**](https://wisp.gg) | Extra features. |
-| [**MixmlHosting**](https://mixmlhosting.com) | MixmlHosting provides high quality Virtual Private Servers along with game servers, all at a affordable price. |
 | [**BisectHosting**](https://www.bisecthosting.com/) | BisectHosting provides Minecraft, Valheim and other server hosting services with the highest reliability and lightning fast support since 2012. |
+| [**Tempest**](https://tempest.net/) | Tempest Hosting is a subsidiary of Path Network, Inc. offering unmetered DDoS protected 10Gbps dedicated servers, starting at just $80/month. Full anycast, tons of filters. |
 | [**Bloom.host**](https://bloom.host) | Bloom.host offers dedicated core VPS and Minecraft hosting with Ryzen 9 processors. With owned-hardware, we offer truly unbeatable prices on high-performance hosting. |
-| [**MineStrator**](https://minestrator.com/) | Looking for a French highend hosting company for you minecraft server? More than 14,000 members on our discord, trust us. |
-| [**DedicatedMC**](https://dedicatedmc.io/) | DedicatedMC provides Raw Power hosting at affordable pricing, making sure to never compromise on your performance and giving you the best performance money can buy. |
+| [**MineStrator**](https://minestrator.com/) | Looking for the most highend French hosting company for your minecraft server? More than 24,000 members on our discord trust us. Give us a try! |
 | [**Skynode**](https://www.skynode.pro/) | Skynode provides blazing fast game servers along with a top-notch user experience. Whatever our clients are looking for, we're able to provide it! |
-| [**XCORE**](https://xcore-server.de/) | XCORE offers High-End Servers for hosting and gaming since 2012. Fast, excellent and well-known for eSports Gaming. |
-| [**RoyaleHosting**](https://royalehosting.net/) | Build your dreams and deploy them with RoyaleHosting’s reliable servers and network. Easy to use, provisioned in a couple of minutes. |
-| [**Spill Hosting**](https://spillhosting.no/) | Spill Hosting is a Norwegian hosting service, which aims for inexpensive services on quality servers. Premium i9-9900K processors will run your game like a dream. |
 | [**DeinServerHost**](https://deinserverhost.de/) | DeinServerHost offers Dedicated, vps and Gameservers for many popular Games like Minecraft and Rust in Germany since 2013. |
-| [**HostBend**](https://hostbend.com/) | HostBend offers a variety of solutions for developers, students, and others who have a tight budget but don't want to compromise quality and support. |
-| [**Capitol Hosting Solutions**](https://chs.gg/) | CHS is *the* budget friendly hosting company for Australian and American gamers, offering a variety of plans from Web Hosting to Game Servers; Custom Solutions too! |
-| [**ByteAnia**](https://byteania.com/?utm_source=pterodactyl) | ByteAnia offers the best performing and most affordable **Ryzen 5000 Series hosting** on the market for *unbeatable prices*! |
 | [**Aussie Server Hosts**](https://aussieserverhosts.com/) | No frills Australian Owned and operated High Performance Server hosting for some of the most demanding games serving Australia and New Zealand. |
+| [**HostEZ**](https://hostez.io) | Providing North America Valheim, Minecraft and other popular games with low latency, high uptime and maximum availability. EZ! |
 | [**VibeGAMES**](https://vibegames.net/) | VibeGAMES is a game server provider that specializes in DDOS protection for the games we offer. We have multiple locations in the US, Brazil, France, Germany, Singapore, Australia and South Africa.|
-| [**RocketNode**](https://rocketnode.net) | RocketNode is a VPS and Game Server provider that offers the best performing VPS and Game hosting Solutions at affordable prices! |
+| [**Gamenodes**](https://gamenodes.nl) | Gamenodes love quality. For Minecraft, Discord Bots and other services, among others. With our own programmers, we provide just that little bit of extra service! |
 
 ## Documentation
 * [Panel Documentation](https://pterodactyl.io/panel/1.0/getting_started.html)

cmd/root.go

@@ -5,12 +5,14 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
+	"github.com/pterodactyl/wings/internal/cron"
 	log2 "log"
 	"net/http"
 	_ "net/http/pprof"
 	"os"
 	"path"
 	"path/filepath"
+	"runtime"
 	"strconv"
 	"strings"
 	"time"
@@ -76,6 +78,7 @@ func init() {
 
 	// Flags specifically used when running the API.
 	rootCommand.Flags().Bool("pprof", false, "if the pprof profiler should be enabled. The profiler will bind to localhost:6060 by default")
+	rootCommand.Flags().Int("pprof-block-rate", 0, "enables block profile support, may have performance impacts")
 	rootCommand.Flags().Int("pprof-port", 6060, "If provided with --pprof, the port it will run on")
 	rootCommand.Flags().Bool("auto-tls", false, "pass in order to have wings generate and manage it's own SSL certificates using Let's Encrypt")
 	rootCommand.Flags().String("tls-hostname", "", "required with --auto-tls, the FQDN for the generated SSL certificate")
@@ -257,6 +260,13 @@ func rootCmdRun(cmd *cobra.Command, _ []string) {
 		}
 	}()
 
+	if s, err := cron.Scheduler(manager); err != nil {
+		log.WithField("error", err).Fatal("failed to initialize cron system")
+	} else {
+		log.WithField("subsystem", "cron").Info("starting cron processes")
+		s.StartAsync()
+	}
+
 	go func() {
 		// Run the SFTP server.
 		if err := sftp.New(manager).Run(); err != nil {
@@ -309,6 +319,12 @@ func rootCmdRun(cmd *cobra.Command, _ []string) {
 
 	profile, _ := cmd.Flags().GetBool("pprof")
 	if profile {
+		if r, _ := cmd.Flags().GetInt("pprof-block-rate"); r > 0 {
+			runtime.SetBlockProfileRate(r)
+		}
+		// Catch at least 1% of mutex contention issues.
+		runtime.SetMutexProfileFraction(100)
+
 		profilePort, _ := cmd.Flags().GetInt("pprof-port")
 		go func() {
 			http.ListenAndServe(fmt.Sprintf("localhost:%d", profilePort), nil)

config/config.go

@@ -89,8 +89,8 @@ type ApiConfiguration struct {
 	// servers.
 	DisableRemoteDownload bool `json:"disable_remote_download" yaml:"disable_remote_download"`
 
-	// The maximum size for files uploaded through the Panel in bytes.
-	UploadLimit int `default:"100" json:"upload_limit" yaml:"upload_limit"`
+	// The maximum size for files uploaded through the Panel in MB.
+	UploadLimit int64 `default:"100" json:"upload_limit" yaml:"upload_limit"`
 }
 
 // RemoteQueryConfiguration defines the configuration settings for remote requests
@@ -132,6 +132,10 @@ type SystemConfiguration struct {
 	// Directory where local backups will be stored on the machine.
 	BackupDirectory string `default:"/var/lib/pterodactyl/backups" yaml:"backup_directory"`
 
+	// TmpDirectory specifies where temporary files for Pterodactyl installation processes
+	// should be created. This supports environments running docker-in-docker.
+	TmpDirectory string `default:"/tmp/pterodactyl" yaml:"tmp_directory"`
+
 	// The user that should own all of the server files, and be used for containers.
 	Username string `default:"pterodactyl" yaml:"username"`
 
@@ -159,6 +163,15 @@ type SystemConfiguration struct {
 	// disk usage is not a concern.
 	DiskCheckInterval int64 `default:"150" yaml:"disk_check_interval"`
 
+	// ActivitySendInterval is the amount of time that should ellapse between aggregated server activity
+	// being sent to the Panel. By default this will send activity collected over the last minute. Keep
+	// in mind that only a fixed number of activity log entries, defined by ActivitySendCount, will be sent
+	// in each run.
+	ActivitySendInterval int64 `default:"60" yaml:"activity_send_interval"`
+
+	// ActivitySendCount is the number of activity events to send per batch.
+	ActivitySendCount int64 `default:"100" yaml:"activity_send_count"`
+
 	// If set to true, file permissions for a server will be checked when the process is
 	// booted. This can cause boot delays if the server has a large amount of files. In most
 	// cases disabling this should not have any major impact unless external processes are

environment/docker/api.go

@@ -73,6 +73,9 @@ func (e *Environment) ContainerInspect(ctx context.Context) (types.ContainerJSON
 
 	res, err := e.client.HTTPClient().Do(req)
 	if err != nil {
+		if res == nil {
+			return st, errdefs.Unknown(err)
+		}
 		return st, errdefs.FromStatusCode(err, res.StatusCode)
 	}
 

environment/docker/container.go

@@ -480,21 +480,3 @@ func (e *Environment) convertMounts() []mount.Mount {
 
 	return out
 }
-
-func (e *Environment) resources() container.Resources {
-	l := e.Configuration.Limits()
-	pids := l.ProcessLimit()
-
-	return container.Resources{
-		Memory:            l.BoundedMemoryLimit(),
-		MemoryReservation: l.MemoryLimit * 1_000_000,
-		MemorySwap:        l.ConvertedSwap(),
-		CPUQuota:          l.ConvertedCpuLimit(),
-		CPUPeriod:         100_000,
-		CPUShares:         1024,
-		BlkioWeight:       l.IoWeight,
-		OomKillDisable:    &l.OOMDisabled,
-		CpusetCpus:        l.Threads,
-		PidsLimit:         &pids,
-	}
-}

environment/docker/environment.go

@@ -26,7 +26,7 @@ type Metadata struct {
 var _ environment.ProcessEnvironment = (*Environment)(nil)
 
 type Environment struct {
-	mu      sync.RWMutex
+	mu sync.RWMutex
 
 	// The public identifier for this environment. In this case it is the Docker container
 	// name that will be used for all instances created under it.

environment/docker/power.go

@@ -138,9 +138,7 @@ func (e *Environment) Start(ctx context.Context) error {
 // You most likely want to be using WaitForStop() rather than this function,
 // since this will return as soon as the command is sent, rather than waiting
 // for the process to be completed stopped.
-//
-// TODO: pass context through from the server instance.
-func (e *Environment) Stop() error {
+func (e *Environment) Stop(ctx context.Context) error {
 	e.mu.RLock()
 	s := e.meta.Stop
 	e.mu.RUnlock()
@@ -164,7 +162,7 @@ func (e *Environment) Stop() error {
 		case "SIGTERM":
 			signal = syscall.SIGTERM
 		}
-		return e.Terminate(signal)
+		return e.Terminate(ctx, signal)
 	}
 
 	// If the process is already offline don't switch it back to stopping. Just leave it how
@@ -179,8 +177,10 @@ func (e *Environment) Stop() error {
 		return e.SendCommand(s.Value)
 	}
 
-	t := time.Second * 30
-	if err := e.client.ContainerStop(context.Background(), e.Id, &t); err != nil {
+	// Allow the stop action to run for however long it takes, similar to executing a command
+	// and using a different logic pathway to wait for the container to stop successfully.
+	t := time.Duration(-1)
+	if err := e.client.ContainerStop(ctx, e.Id, &t); err != nil {
 		// If the container does not exist just mark the process as stopped and return without
 		// an error.
 		if client.IsErrNotFound(err) {
@@ -198,45 +198,66 @@ func (e *Environment) Stop() error {
 // command. If the server does not stop after seconds have passed, an error will
 // be returned, or the instance will be terminated forcefully depending on the
 // value of the second argument.
-func (e *Environment) WaitForStop(seconds uint, terminate bool) error {
-	if err := e.Stop(); err != nil {
-		return err
+//
+// Calls to Environment.Terminate() in this function use the context passed
+// through since we don't want to prevent termination of the server instance
+// just because the context.WithTimeout() has expired.
+func (e *Environment) WaitForStop(ctx context.Context, duration time.Duration, terminate bool) error {
+	tctx, cancel := context.WithTimeout(context.Background(), duration)
+	defer cancel()
+
+	// If the parent context is canceled, abort the timed context for termination.
+	go func() {
+		select {
+		case <-ctx.Done():
+			cancel()
+		case <-tctx.Done():
+			// When the timed context is canceled, terminate this routine since we no longer
+			// need to worry about the parent routine being canceled.
+			break
+		}
+	}()
+
+	doTermination := func(s string) error {
+		e.log().WithField("step", s).WithField("duration", duration).Warn("container stop did not complete in time, terminating process...")
+		return e.Terminate(ctx, os.Kill)
 	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(seconds)*time.Second)
-	defer cancel()
+	// We pass through the timed context for this stop action so that if one of the
+	// internal docker calls fails to ever finish before we've exhausted the time limit
+	// the resources get cleaned up, and the exection is stopped.
+	if err := e.Stop(tctx); err != nil {
+		if terminate && errors.Is(err, context.DeadlineExceeded) {
+			return doTermination("stop")
+		}
+		return err
+	}
 
 	// Block the return of this function until the container as been marked as no
 	// longer running. If this wait does not end by the time seconds have passed,
 	// attempt to terminate the container, or return an error.
-	ok, errChan := e.client.ContainerWait(ctx, e.Id, container.WaitConditionNotRunning)
+	ok, errChan := e.client.ContainerWait(tctx, e.Id, container.WaitConditionNotRunning)
 	select {
 	case <-ctx.Done():
-		if ctxErr := ctx.Err(); ctxErr != nil {
+		if err := ctx.Err(); err != nil {
 			if terminate {
-				log.WithField("container_id", e.Id).Info("server did not stop in time, executing process termination")
-
-				return e.Terminate(os.Kill)
+				return doTermination("parent-context")
 			}
-
-			return ctxErr
+			return err
 		}
 	case err := <-errChan:
 		// If the error stems from the container not existing there is no point in wasting
 		// CPU time to then try and terminate it.
-		if err != nil && !client.IsErrNotFound(err) {
-			if terminate {
-				l := log.WithField("container_id", e.Id)
-				if errors.Is(err, context.DeadlineExceeded) {
-					l.Warn("deadline exceeded for container stop; terminating process")
-				} else {
-					l.WithField("error", err).Warn("error while waiting for container stop; terminating process")
-				}
-
-				return e.Terminate(os.Kill)
-			}
-			return errors.WrapIf(err, "environment/docker: error waiting on container to enter \"not-running\" state")
+		if err == nil || client.IsErrNotFound(err) {
+			return nil
 		}
+		if terminate {
+			if !errors.Is(err, context.DeadlineExceeded) {
+				e.log().WithField("error", err).Warn("error while waiting for container stop; terminating process")
+			}
+			return doTermination("wait")
+		}
+		return errors.WrapIf(err, "environment/docker: error waiting on container to enter \"not-running\" state")
 	case <-ok:
 	}
 
@@ -244,8 +265,8 @@ func (e *Environment) WaitForStop(seconds uint, terminate bool) error {
 }
 
 // Terminate forcefully terminates the container using the signal provided.
-func (e *Environment) Terminate(signal os.Signal) error {
-	c, err := e.ContainerInspect(context.Background())
+func (e *Environment) Terminate(ctx context.Context, signal os.Signal) error {
+	c, err := e.ContainerInspect(ctx)
 	if err != nil {
 		// Treat missing containers as an okay error state, means it is obviously
 		// already terminated at this point.
@@ -270,7 +291,7 @@ func (e *Environment) Terminate(signal os.Signal) error {
 	// We set it to stopping than offline to prevent crash detection from being triggered.
 	e.SetState(environment.ProcessStoppingState)
 	sig := strings.TrimSuffix(strings.TrimPrefix(signal.String(), "signal "), "ed")
-	if err := e.client.ContainerKill(context.Background(), e.Id, sig); err != nil && !client.IsErrNotFound(err) {
+	if err := e.client.ContainerKill(ctx, e.Id, sig); err != nil && !client.IsErrNotFound(err) {
 		return errors.WithStack(err)
 	}
 	e.SetState(environment.ProcessOfflineState)

environment/environment.go

@@ -3,6 +3,7 @@ package environment
 import (
 	"context"
 	"os"
+	"time"
 
 	"github.com/pterodactyl/wings/events"
 )
@@ -58,18 +59,20 @@ type ProcessEnvironment interface {
 	// can be started an error should be returned.
 	Start(ctx context.Context) error
 
-	// Stops a server instance. If the server is already stopped an error should
-	// not be returned.
-	Stop() error
+	// Stop stops a server instance. If the server is already stopped an error will
+	// not be returned, this function will act as a no-op.
+	Stop(ctx context.Context) error
 
-	// Waits for a server instance to stop gracefully. If the server is still detected
-	// as running after seconds, an error will be returned, or the server will be terminated
-	// depending on the value of the second argument.
-	WaitForStop(seconds uint, terminate bool) error
+	// WaitForStop waits for a server instance to stop gracefully. If the server is
+	// still detected as running after "duration", an error will be returned, or the server
+	// will be terminated depending on the value of the second argument. If the context
+	// provided is canceled the underlying wait conditions will be stopped and the
+	// entire loop will be ended (potentially without stopping or terminating).
+	WaitForStop(ctx context.Context, duration time.Duration, terminate bool) error
 
-	// Terminates a running server instance using the provided signal. If the server
-	// is not running no error should be returned.
-	Terminate(signal os.Signal) error
+	// Terminate stops a running server instance using the provided signal. This function
+	// is a no-op if the server is already stopped.
+	Terminate(ctx context.Context, signal os.Signal) error
 
 	// Destroys the environment removing any containers that were created (in Docker
 	// environments at least).

environment/settings.go

@@ -99,21 +99,36 @@ func (l Limits) ProcessLimit() int64 {
 	return config.Get().Docker.ContainerPidLimit
 }
 
+// AsContainerResources returns the available resources for a container in a format
+// that Docker understands.
 func (l Limits) AsContainerResources() container.Resources {
 	pids := l.ProcessLimit()
-
-	return container.Resources{
+	resources := container.Resources{
 		Memory:            l.BoundedMemoryLimit(),
 		MemoryReservation: l.MemoryLimit * 1_000_000,
 		MemorySwap:        l.ConvertedSwap(),
-		CPUQuota:          l.ConvertedCpuLimit(),
-		CPUPeriod:         100_000,
-		CPUShares:         1024,
 		BlkioWeight:       l.IoWeight,
 		OomKillDisable:    &l.OOMDisabled,
-		CpusetCpus:        l.Threads,
 		PidsLimit:         &pids,
 	}
+
+	// If the CPU Limit is not set, don't send any of these fields through. Providing
+	// them seems to break some Java services that try to read the available processors.
+	//
+	// @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
+	}
+
+	// Similar to above, don't set the specific assigned CPUs if we didn't actually limit
+	// the server to any of them.
+	if l.Threads != "" {
+		resources.CpusetCpus = l.Threads
+	}
+
+	return resources
 }
 
 type Variables map[string]interface{}

events/events.go

@@ -2,10 +2,11 @@ package events
 
 import (
 	"strings"
-	"sync"
-)
 
-type Listener chan Event
+	"emperror.dev/errors"
+	"github.com/goccy/go-json"
+	"github.com/pterodactyl/wings/system"
+)
 
 // Event represents an Event sent over a Bus.
 type Event struct {
@@ -15,137 +16,55 @@ type Event struct {
 
 // Bus represents an Event Bus.
 type Bus struct {
-	listenersMx sync.Mutex
-	listeners   map[string][]Listener
+	*system.SinkPool
 }
 
-// NewBus returns a new empty Event Bus.
+// NewBus returns a new empty Bus. This is simply a nicer wrapper around the
+// system.SinkPool implementation that allows for more simplistic usage within
+// the codebase.
+//
+// All of the events emitted out of this bus are byte slices that can be decoded
+// back into an events.Event interface.
 func NewBus() *Bus {
 	return &Bus{
-		listeners: make(map[string][]Listener),
-	}
-}
-
-// Off unregisters a listener from the specified topics on the Bus.
-func (b *Bus) Off(listener Listener, topics ...string) {
-	b.listenersMx.Lock()
-	defer b.listenersMx.Unlock()
-
-	var closed bool
-
-	for _, topic := range topics {
-		ok := b.off(topic, listener)
-		if !closed && ok {
-			close(listener)
-			closed = true
-		}
-	}
-}
-
-func (b *Bus) off(topic string, listener Listener) bool {
-	listeners, ok := b.listeners[topic]
-	if !ok {
-		return false
-	}
-	for i, l := range listeners {
-		if l != listener {
-			continue
-		}
-
-		listeners = append(listeners[:i], listeners[i+1:]...)
-		b.listeners[topic] = listeners
-		return true
-	}
-	return false
-}
-
-// On registers a listener to the specified topics on the Bus.
-func (b *Bus) On(listener Listener, topics ...string) {
-	b.listenersMx.Lock()
-	defer b.listenersMx.Unlock()
-
-	for _, topic := range topics {
-		b.on(topic, listener)
-	}
-}
-
-func (b *Bus) on(topic string, listener Listener) {
-	listeners, ok := b.listeners[topic]
-	if !ok {
-		b.listeners[topic] = []Listener{listener}
-	} else {
-		b.listeners[topic] = append(listeners, listener)
+		system.NewSinkPool(),
 	}
 }
 
 // Publish publishes a message to the Bus.
 func (b *Bus) Publish(topic string, data interface{}) {
-	// Some of our topics for the socket support passing a more specific namespace,
+	// Some of our actions for the socket support passing a more specific namespace,
 	// such as "backup completed:1234" to indicate which specific backup was completed.
 	//
 	// In these cases, we still need to send the event using the standard listener
 	// name of "backup completed".
 	if strings.Contains(topic, ":") {
 		parts := strings.SplitN(topic, ":", 2)
-
 		if len(parts) == 2 {
 			topic = parts[0]
 		}
 	}
 
-	b.listenersMx.Lock()
-	defer b.listenersMx.Unlock()
-
-	listeners, ok := b.listeners[topic]
-	if !ok {
-		return
+	enc, err := json.Marshal(Event{Topic: topic, Data: data})
+	if err != nil {
+		panic(errors.WithStack(err))
 	}
-	if len(listeners) < 1 {
-		return
-	}
-
-	var wg sync.WaitGroup
-	event := Event{Topic: topic, Data: data}
-	for _, listener := range listeners {
-		l := listener
-		wg.Add(1)
-		go func(l Listener, event Event) {
-			defer wg.Done()
-			l <- event
-		}(l, event)
-	}
-	wg.Wait()
+	b.Push(enc)
 }
 
-// Destroy destroys the Event Bus by unregistering and closing all listeners.
-func (b *Bus) Destroy() {
-	b.listenersMx.Lock()
-	defer b.listenersMx.Unlock()
-
-	// Track what listeners have already been closed. Because the same listener
-	// can be listening on multiple topics, we need a way to essentially
-	// "de-duplicate" all the listeners across all the topics.
-	var closed []Listener
-
-	for _, listeners := range b.listeners {
-		for _, listener := range listeners {
-			if contains(closed, listener) {
-				continue
-			}
-
-			close(listener)
-			closed = append(closed, listener)
-		}
+// MustDecode decodes the event byte slice back into an events.Event struct or
+// panics if an error is encountered during this process.
+func MustDecode(data []byte) (e Event) {
+	if err := DecodeTo(data, &e); err != nil {
+		panic(err)
 	}
-
-	b.listeners = make(map[string][]Listener)
+	return
 }
 
-func contains(closed []Listener, listener Listener) bool {
-	for _, c := range closed {
-		if c == listener {
-			return true
-		}
+// DecodeTo decodes a byte slice of event data into the given interface.
+func DecodeTo(data []byte, v interface{}) error {
+	if err := json.Unmarshal(data, &v); err != nil {
+		return errors.Wrap(err, "events: failed to decode byte slice")
 	}
-	return false
+	return nil
 }

events/events_test.go

@@ -9,162 +9,90 @@ import (
 
 func TestNewBus(t *testing.T) {
 	g := Goblin(t)
-	bus := NewBus()
 
-	g.Describe("NewBus", func() {
-		g.It("is not nil", func() {
-			g.Assert(bus).IsNotNil("Bus expected to not be nil")
-			g.Assert(bus.listeners).IsNotNil("Bus#listeners expected to not be nil")
-		})
-	})
-}
-
-func TestBus_Off(t *testing.T) {
-	g := Goblin(t)
-
-	const topic = "test"
-
-	g.Describe("Off", func() {
-		g.It("unregisters listener", func() {
-			bus := NewBus()
-
-			g.Assert(bus.listeners[topic]).IsNotNil()
-			g.Assert(len(bus.listeners[topic])).IsZero()
-			listener := make(chan Event)
-			bus.On(listener, topic)
-			g.Assert(len(bus.listeners[topic])).Equal(1, "Listener was not registered")
-
-			bus.Off(listener, topic)
-			g.Assert(len(bus.listeners[topic])).Equal(0, "Topic still has one or more listeners")
+	g.Describe("Events", func() {
+		var bus *Bus
+		g.BeforeEach(func() {
+			bus = NewBus()
 		})
 
-		g.It("unregisters correct listener", func() {
-			bus := NewBus()
-
-			listener := make(chan Event)
-			listener2 := make(chan Event)
-			listener3 := make(chan Event)
-			bus.On(listener, topic)
-			bus.On(listener2, topic)
-			bus.On(listener3, topic)
-			g.Assert(len(bus.listeners[topic])).Equal(3, "Listeners were not registered")
-
-			bus.Off(listener, topic)
-			bus.Off(listener3, topic)
-			g.Assert(len(bus.listeners[topic])).Equal(1, "Expected 1 listener to remain")
-
-			if bus.listeners[topic][0] != listener2 {
-				// A normal Assert does not properly compare channels.
-				g.Fail("wrong listener unregistered")
-			}
-
-			// Cleanup
-			bus.Off(listener2, topic)
-		})
-	})
-}
-
-func TestBus_On(t *testing.T) {
-	g := Goblin(t)
-
-	const topic = "test"
-
-	g.Describe("On", func() {
-		g.It("registers listener", func() {
-			bus := NewBus()
-
-			g.Assert(bus.listeners[topic]).IsNotNil()
-			g.Assert(len(bus.listeners[topic])).IsZero()
-			listener := make(chan Event)
-			bus.On(listener, topic)
-			g.Assert(len(bus.listeners[topic])).Equal(1, "Listener was not registered")
-
-			if bus.listeners[topic][0] != listener {
-				// A normal Assert does not properly compare channels.
-				g.Fail("wrong listener registered")
-			}
-
-			// Cleanup
-			bus.Off(listener, topic)
-		})
-	})
-}
-
-func TestBus_Publish(t *testing.T) {
-	g := Goblin(t)
-
-	const topic = "test"
-	const message = "this is a test message!"
-
-	g.Describe("Publish", func() {
-		g.It("publishes message", func() {
-			bus := NewBus()
-
-			g.Assert(bus.listeners[topic]).IsNotNil()
-			g.Assert(len(bus.listeners[topic])).IsZero()
-			listener := make(chan Event)
-			bus.On(listener, topic)
-			g.Assert(len(bus.listeners[topic])).Equal(1, "Listener was not registered")
-
-			done := make(chan struct{}, 1)
-			go func() {
-				select {
-				case m := <-listener:
-					g.Assert(m.Topic).Equal(topic)
-					g.Assert(m.Data).Equal(message)
-				case <-time.After(1 * time.Second):
-					g.Fail("listener did not receive message in time")
-				}
-				done <- struct{}{}
-			}()
-			bus.Publish(topic, message)
-			<-done
-
-			// Cleanup
-			bus.Off(listener, topic)
+		g.Describe("NewBus", func() {
+			g.It("is not nil", func() {
+				g.Assert(bus).IsNotNil("Bus expected to not be nil")
+			})
 		})
 
-		g.It("publishes message to all listeners", func() {
-			bus := NewBus()
+		g.Describe("Publish", func() {
+			const topic = "test"
+			const message = "this is a test message!"
 
-			g.Assert(bus.listeners[topic]).IsNotNil()
-			g.Assert(len(bus.listeners[topic])).IsZero()
-			listener := make(chan Event)
-			listener2 := make(chan Event)
-			listener3 := make(chan Event)
-			bus.On(listener, topic)
-			bus.On(listener2, topic)
-			bus.On(listener3, topic)
-			g.Assert(len(bus.listeners[topic])).Equal(3, "Listener was not registered")
+			g.It("publishes message", func() {
+				bus := NewBus()
 
-			done := make(chan struct{}, 1)
-			go func() {
-				for i := 0; i < 3; i++ {
+				listener := make(chan []byte)
+				bus.On(listener)
+
+				done := make(chan struct{}, 1)
+				go func() {
 					select {
-					case m := <-listener:
-						g.Assert(m.Topic).Equal(topic)
-						g.Assert(m.Data).Equal(message)
-					case m := <-listener2:
-						g.Assert(m.Topic).Equal(topic)
-						g.Assert(m.Data).Equal(message)
-					case m := <-listener3:
+					case v := <-listener:
+						m := MustDecode(v)
 						g.Assert(m.Topic).Equal(topic)
 						g.Assert(m.Data).Equal(message)
 					case <-time.After(1 * time.Second):
-						g.Fail("all listeners did not receive the message in time")
-						i = 3
+						g.Fail("listener did not receive message in time")
 					}
-				}
+					done <- struct{}{}
+				}()
+				bus.Publish(topic, message)
+				<-done
 
-				done <- struct{}{}
-			}()
-			bus.Publish(topic, message)
-			<-done
+				// Cleanup
+				bus.Off(listener)
+			})
 
-			// Cleanup
-			bus.Off(listener, topic)
-			bus.Off(listener2, topic)
-			bus.Off(listener3, topic)
+			g.It("publishes message to all listeners", func() {
+				bus := NewBus()
+
+				listener := make(chan []byte)
+				listener2 := make(chan []byte)
+				listener3 := make(chan []byte)
+				bus.On(listener)
+				bus.On(listener2)
+				bus.On(listener3)
+
+				done := make(chan struct{}, 1)
+				go func() {
+					for i := 0; i < 3; i++ {
+						select {
+						case v := <-listener:
+							m := MustDecode(v)
+							g.Assert(m.Topic).Equal(topic)
+							g.Assert(m.Data).Equal(message)
+						case v := <-listener2:
+							m := MustDecode(v)
+							g.Assert(m.Topic).Equal(topic)
+							g.Assert(m.Data).Equal(message)
+						case v := <-listener3:
+							m := MustDecode(v)
+							g.Assert(m.Topic).Equal(topic)
+							g.Assert(m.Data).Equal(message)
+						case <-time.After(1 * time.Second):
+							g.Fail("all listeners did not receive the message in time")
+							i = 3
+						}
+					}
+
+					done <- struct{}{}
+				}()
+				bus.Publish(topic, message)
+				<-done
+
+				// Cleanup
+				bus.Off(listener)
+				bus.Off(listener2)
+				bus.Off(listener3)
+			})
 		})
 	})
 }

go.mod

@@ -3,116 +3,119 @@ module github.com/pterodactyl/wings
 go 1.17
 
 require (
-	emperror.dev/errors v0.8.0
-	github.com/AlecAivazis/survey/v2 v2.2.15
+	emperror.dev/errors v0.8.1
+	github.com/AlecAivazis/survey/v2 v2.3.4
 	github.com/Jeffail/gabs/v2 v2.6.1
 	github.com/NYTimes/logrotate v1.0.0
 	github.com/apex/log v1.9.0
 	github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d
 	github.com/beevik/etree v1.1.0
 	github.com/buger/jsonparser v1.1.1
-	github.com/cenkalti/backoff/v4 v4.1.1
+	github.com/cenkalti/backoff/v4 v4.1.2
 	github.com/cobaugh/osrelease v0.0.0-20181218015638-a93a0a55a249
-	github.com/creasty/defaults v1.5.1
-	github.com/docker/docker v20.10.7+incompatible
+	github.com/creasty/defaults v1.5.2
+	github.com/docker/docker v20.10.14+incompatible
 	github.com/docker/go-connections v0.4.0
-	github.com/fatih/color v1.12.0
+	github.com/fatih/color v1.13.0
 	github.com/franela/goblin v0.0.0-20200825194134-80c0062ed6cd
-	github.com/gabriel-vasile/mimetype v1.3.1
+	github.com/gabriel-vasile/mimetype v1.4.0
 	github.com/gammazero/workerpool v1.1.2
 	github.com/gbrlsnchs/jwt/v3 v3.0.1
-	github.com/gin-gonic/gin v1.7.2
+	github.com/gin-gonic/gin v1.7.7
 	github.com/google/uuid v1.3.0
-	github.com/gorilla/websocket v1.4.2
+	github.com/gorilla/websocket v1.5.0
 	github.com/iancoleman/strcase v0.2.0
 	github.com/icza/dyno v0.0.0-20210726202311-f1bafe5d9996
 	github.com/juju/ratelimit v1.0.1
 	github.com/karrick/godirwalk v1.16.1
 	github.com/klauspost/pgzip v1.2.5
-	github.com/magiconair/properties v1.8.5
-	github.com/mattn/go-colorable v0.1.8
-	github.com/mholt/archiver/v3 v3.5.0
+	github.com/magiconair/properties v1.8.6
+	github.com/mattn/go-colorable v0.1.12
+	github.com/mholt/archiver/v3 v3.5.1
 	github.com/mitchellh/colorstring v0.0.0-20190213212951-d06e56a500db
 	github.com/patrickmn/go-cache v2.1.0+incompatible
-	github.com/pkg/profile v1.6.0
-	github.com/pkg/sftp v1.13.2
-	github.com/sabhiram/go-gitignore v0.0.0-20201211210132-54b8a0bf510f
-	github.com/spf13/cobra v1.2.1
-	github.com/stretchr/testify v1.7.0
-	golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97
+	github.com/pkg/sftp v1.13.4
+	github.com/sabhiram/go-gitignore v0.0.0-20210923224102-525f6e181f06
+	github.com/spf13/cobra v1.4.0
+	github.com/stretchr/testify v1.7.5
+	golang.org/x/crypto v0.0.0-20220321153916-2c7772ba3064
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
-	gopkg.in/ini.v1 v1.62.0
+	gopkg.in/ini.v1 v1.66.4
 	gopkg.in/yaml.v2 v2.4.0
 )
 
-require github.com/goccy/go-json v0.9.4
+require github.com/goccy/go-json v0.9.6
 
-require golang.org/x/sys v0.0.0-20211110154304-99a53858aa08 // indirect
+require golang.org/x/sys v0.0.0-20220704084225-05e143d24a9e // indirect
 
 require (
 	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
-	github.com/Microsoft/go-winio v0.5.0 // indirect
-	github.com/Microsoft/hcsshim v0.8.20 // indirect
-	github.com/andybalholm/brotli v1.0.3 // indirect
+	github.com/Microsoft/go-winio v0.5.2 // indirect
+	github.com/Microsoft/hcsshim v0.9.2 // indirect
+	github.com/andybalholm/brotli v1.0.4 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
-	github.com/cespare/xxhash/v2 v2.1.1 // indirect
-	github.com/containerd/containerd v1.5.5 // indirect
+	github.com/bwmarrin/snowflake v0.3.0 // indirect
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/containerd/containerd v1.6.2 // indirect
 	github.com/containerd/fifo v1.0.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/docker/distribution v2.7.1+incompatible // indirect
+	github.com/docker/distribution v2.8.1+incompatible // indirect
 	github.com/docker/go-metrics v0.0.1 // indirect
 	github.com/docker/go-units v0.4.0 // indirect
-	github.com/dsnet/compress v0.0.1 // indirect
-	github.com/fsnotify/fsnotify v1.4.9 // indirect
-	github.com/gammazero/deque v0.1.0 // indirect
+	github.com/dsnet/compress v0.0.2-0.20210315054119-f66993602bf5 // indirect
+	github.com/fsnotify/fsnotify v1.5.1 // indirect
+	github.com/gammazero/deque v0.1.1 // indirect
 	github.com/gin-contrib/sse v0.1.0 // indirect
-	github.com/go-playground/locales v0.13.0 // indirect
-	github.com/go-playground/universal-translator v0.17.0 // indirect
-	github.com/go-playground/validator/v10 v10.8.0 // indirect
+	github.com/go-co-op/gocron v1.15.0 // indirect
+	github.com/go-playground/locales v0.14.0 // indirect
+	github.com/go-playground/universal-translator v0.18.0 // indirect
+	github.com/go-playground/validator/v10 v10.10.1 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/gorilla/mux v1.7.4 // indirect
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect
-	github.com/json-iterator/go v1.1.11 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 // indirect
-	github.com/klauspost/compress v1.13.2 // indirect
+	github.com/klauspost/compress v1.15.1 // indirect
 	github.com/kr/fs v0.1.0 // indirect
 	github.com/leodido/go-urn v1.2.1 // indirect
-	github.com/magefile/mage v1.11.0 // indirect
-	github.com/mattn/go-isatty v0.0.13 // indirect
+	github.com/magefile/mage v1.13.0 // indirect
+	github.com/mattn/go-isatty v0.0.14 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
 	github.com/mgutz/ansi v0.0.0-20200706080929-d51e80ef957d // indirect
 	github.com/moby/term v0.0.0-20210619224110-3f7ff695adc6 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
-	github.com/modern-go/reflect2 v1.0.1 // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/morikuni/aec v1.0.0 // indirect
-	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect
-	github.com/nwaples/rardecode v1.1.1 // indirect
+	github.com/nwaples/rardecode v1.1.3 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
-	github.com/opencontainers/image-spec v1.0.1 // indirect
-	github.com/pierrec/lz4/v4 v4.1.8 // indirect
+	github.com/opencontainers/image-spec v1.0.2 // indirect
+	github.com/pierrec/lz4/v4 v4.1.14 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/prometheus/client_golang v1.11.0 // indirect
+	github.com/prometheus/client_golang v1.12.1 // indirect
 	github.com/prometheus/client_model v0.2.0 // indirect
-	github.com/prometheus/common v0.30.0 // indirect
-	github.com/prometheus/procfs v0.7.1 // indirect
+	github.com/prometheus/common v0.32.1 // indirect
+	github.com/prometheus/procfs v0.7.3 // indirect
+	github.com/robfig/cron/v3 v3.0.1 // indirect
 	github.com/sirupsen/logrus v1.8.1 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
-	github.com/ugorji/go/codec v1.1.7 // indirect
+	github.com/ugorji/go/codec v1.2.7 // indirect
 	github.com/ulikunitz/xz v0.5.10 // indirect
 	github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect
+	github.com/xujiajun/mmap-go v1.0.1 // indirect
+	github.com/xujiajun/nutsdb v0.9.0 // indirect
+	github.com/xujiajun/utils v0.0.0-20190123093513-8bf096c4f53b // indirect
 	go.uber.org/atomic v1.9.0 // indirect
-	go.uber.org/multierr v1.7.0 // indirect
-	golang.org/x/net v0.0.0-20210726213435-c6fcb2dbf985 // indirect
-	golang.org/x/term v0.0.0-20210615171337-6886f2dfbf5b // indirect
-	golang.org/x/text v0.3.6 // indirect
-	golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac // indirect
+	go.uber.org/multierr v1.8.0 // indirect
+	golang.org/x/net v0.0.0-20220225172249-27dd8689420f // indirect
+	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 // indirect
+	golang.org/x/text v0.3.7 // indirect
+	golang.org/x/time v0.0.0-20220224211638-0e9765cccd65 // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
-	google.golang.org/genproto v0.0.0-20210729151513-df9385d47c1b // indirect
-	google.golang.org/grpc v1.39.0 // indirect
-	google.golang.org/protobuf v1.27.1 // indirect
-	gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f // indirect
-	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
+	google.golang.org/genproto v0.0.0-20220324131243-acbaeb5b85eb // indirect
+	google.golang.org/grpc v1.45.0 // indirect
+	google.golang.org/protobuf v1.28.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

internal/cron/activity_cron.go

@@ -0,0 +1,98 @@
+package cron
+
+import (
+	"context"
+	"emperror.dev/errors"
+	"github.com/apex/log"
+	"github.com/goccy/go-json"
+	"github.com/pterodactyl/wings/internal/database"
+	"github.com/pterodactyl/wings/server"
+	"github.com/pterodactyl/wings/system"
+	"github.com/xujiajun/nutsdb"
+)
+
+var key = []byte("events")
+var processing system.AtomicBool
+
+func processActivityLogs(m *server.Manager, c int64) error {
+	// Don't execute this cron if there is currently one running. Once this task is completed
+	// go ahead and mark it as no longer running.
+	if !processing.SwapIf(true) {
+		return nil
+	}
+	defer processing.Store(false)
+
+	var list [][]byte
+	err := database.DB().View(func(tx *nutsdb.Tx) error {
+		// Grab the oldest 100 activity events that have been logged and send them back to the
+		// Panel for processing. Once completed, delete those events from the database and then
+		// release the lock on this process.
+		end := int(c)
+		if s, err := tx.LSize(database.ServerActivityBucket, key); err != nil {
+			return errors.WithStackIf(err)
+		} else if s < end || s == 0 {
+			if s == 0 {
+				return nil
+			}
+			end = s
+		}
+		l, err := tx.LRange(database.ServerActivityBucket, key, 0, end)
+		if err != nil {
+			// This error is returned when the bucket doesn't exist, which is likely on the
+			// first invocations of Wings since we haven't yet logged any data. There is nothing
+			// that needs to be done if this error occurs.
+			if errors.Is(err, nutsdb.ErrBucket) {
+				return nil
+			}
+			return errors.WithStackIf(err)
+		}
+		list = l
+		return nil
+	})
+
+	if err != nil || len(list) == 0 {
+		return errors.WithStackIf(err)
+	}
+
+	var processed []json.RawMessage
+	for _, l := range list {
+		var v json.RawMessage
+		if err := json.Unmarshal(l, &v); err != nil {
+			log.WithField("error", errors.WithStack(err)).Warn("failed to parse activity event json, skipping entry")
+			continue
+		}
+		processed = append(processed, v)
+	}
+
+	if err := m.Client().SendActivityLogs(context.Background(), processed); err != nil {
+		return errors.WrapIf(err, "cron: failed to send activity events to Panel")
+	}
+
+	return database.DB().Update(func(tx *nutsdb.Tx) error {
+		if m, err := tx.LSize(database.ServerActivityBucket, key); err != nil {
+			return errors.WithStack(err)
+		} else if m > len(list) {
+			// As long as there are more elements than we have in the length of our list
+			// we can just use the existing `LTrim` functionality of nutsdb. This will remove
+			// all of the values we've already pulled and sent to the API.
+			return errors.WithStack(tx.LTrim(database.ServerActivityBucket, key, len(list), -1))
+		} else {
+			i := 0
+			// This is the only way I can figure out to actually empty the items out of the list
+			// because you cannot use `LTrim` (or I cannot for the life of me figure out how) to
+			// trim the slice down to 0 items without it triggering an internal logic error. Perhaps
+			// in a future release they'll have a function to do this (based on my skimming of issues
+			// on GitHub that I cannot read due to translation barriers).
+			for {
+				if i >= m {
+					break
+				}
+				if _, err := tx.LPop(database.ServerActivityBucket, key); err != nil {
+					return errors.WithStack(err)
+				}
+				i++
+			}
+		}
+		return nil
+	})
+}

internal/cron/cron.go

@@ -0,0 +1,36 @@
+package cron
+
+import (
+	"emperror.dev/errors"
+	"github.com/apex/log"
+	"github.com/go-co-op/gocron"
+	"github.com/pterodactyl/wings/config"
+	"github.com/pterodactyl/wings/server"
+	"github.com/pterodactyl/wings/system"
+	"time"
+)
+
+var o system.AtomicBool
+
+// Scheduler configures the internal cronjob system for Wings and returns the scheduler
+// instance to the caller. This should only be called once per application lifecycle, additional
+// calls will result in an error being returned.
+func Scheduler(m *server.Manager) (*gocron.Scheduler, error) {
+	if o.Load() {
+		return nil, errors.New("cron: cannot call scheduler more than once in application lifecycle")
+	}
+	o.Store(true)
+	l, err := time.LoadLocation(config.Get().System.Timezone)
+	if err != nil {
+		return nil, errors.Wrap(err, "cron: failed to parse configured system timezone")
+	}
+
+	s := gocron.NewScheduler(l)
+	_, _ = s.Tag("activity").Every(config.Get().System.ActivitySendInterval).Seconds().Do(func() {
+		if err := processActivityLogs(m, config.Get().System.ActivitySendCount); err != nil {
+			log.WithField("error", err).Error("cron: failed to process activity events")
+		}
+	})
+
+	return s, nil
+}

internal/database/database.go

@@ -0,0 +1,38 @@
+package database
+
+import (
+	"emperror.dev/errors"
+	"github.com/apex/log"
+	"github.com/pterodactyl/wings/config"
+	"github.com/xujiajun/nutsdb"
+	"path/filepath"
+	"sync"
+)
+
+var db *nutsdb.DB
+var syncer sync.Once
+
+const (
+	ServerActivityBucket = "server_activity"
+)
+
+func initialize() error {
+	opt := nutsdb.DefaultOptions
+	opt.Dir = filepath.Join(config.Get().System.RootDirectory, "db")
+
+	instance, err := nutsdb.Open(opt)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+	db = instance
+	return nil
+}
+
+func DB() *nutsdb.DB {
+	syncer.Do(func() {
+		if err := initialize(); err != nil {
+			log.WithField("error", err).Fatal("database: failed to initialize instance, this is an unrecoverable error")
+		}
+	})
+	return db
+}

parser/parser.go

@@ -11,9 +11,9 @@ import (
 	"github.com/apex/log"
 	"github.com/beevik/etree"
 	"github.com/buger/jsonparser"
+	"github.com/goccy/go-json"
 	"github.com/icza/dyno"
 	"github.com/magiconair/properties"
-	"github.com/goccy/go-json"
 	"gopkg.in/ini.v1"
 	"gopkg.in/yaml.v2"
 

remote/http.go

@@ -30,6 +30,7 @@ type Client interface {
 	SetInstallationStatus(ctx context.Context, uuid string, successful bool) error
 	SetTransferStatus(ctx context.Context, uuid string, successful bool) error
 	ValidateSftpCredentials(ctx context.Context, request SftpAuthRequest) (SftpAuthResponse, error)
+	SendActivityLogs(ctx context.Context, activity []json.RawMessage) error
 }
 
 type client struct {
@@ -128,10 +129,19 @@ func (c *client) requestOnce(ctx context.Context, method, path string, body io.R
 // and adds the required authentication headers to the request that is being
 // created. Errors returned will be of the RequestError type if there was some
 // type of response from the API that can be parsed.
-func (c *client) request(ctx context.Context, method, path string, body io.Reader, opts ...func(r *http.Request)) (*Response, error) {
+func (c *client) request(ctx context.Context, method, path string, body *bytes.Buffer, opts ...func(r *http.Request)) (*Response, error) {
 	var res *Response
 	err := backoff.Retry(func() error {
-		r, err := c.requestOnce(ctx, method, path, body, opts...)
+		var b bytes.Buffer
+		if body != nil {
+			// We have to create a copy of the body, otherwise attempting this request again will
+			// send no data if there was initially a body since the "requestOnce" method will read
+			// the whole buffer, thus leaving it empty at the end.
+			if _, err := b.Write(body.Bytes()); err != nil {
+				return backoff.Permanent(errors.Wrap(err, "http: failed to copy body buffer"))
+			}
+		}
+		r, err := c.requestOnce(ctx, method, path, &b, opts...)
 		if err != nil {
 			if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
 				return backoff.Permanent(err)
@@ -142,12 +152,10 @@ func (c *client) request(ctx context.Context, method, path string, body io.Reade
 		if r.HasError() {
 			// Close the request body after returning the error to free up resources.
 			defer r.Body.Close()
-			// Don't keep spamming the endpoint if we've already made too many requests or
-			// if we're not even authenticated correctly. Retrying generally won't fix either
-			// of these issues.
-			if r.StatusCode == http.StatusForbidden ||
-				r.StatusCode == http.StatusTooManyRequests ||
-				r.StatusCode == http.StatusUnauthorized {
+			// Don't keep attempting to access this endpoint if the response is a 4XX
+			// level error which indicates a client mistake. Only retry when the error
+			// is due to a server issue (5XX error).
+			if r.StatusCode >= 400 && r.StatusCode < 500 {
 				return backoff.Permanent(r.Error())
 			}
 			return r.Error()

remote/servers.go

@@ -3,6 +3,7 @@ package remote
 import (
 	"context"
 	"fmt"
+	"github.com/goccy/go-json"
 	"strconv"
 	"sync"
 
@@ -178,6 +179,16 @@ func (c *client) SendRestorationStatus(ctx context.Context, backup string, succe
 	return nil
 }
 
+// SendActivityLogs sends activity logs back to the Panel for processing.
+func (c *client) SendActivityLogs(ctx context.Context, activity []json.RawMessage) error {
+	resp, err := c.Post(ctx, "/activity", d{"data": activity})
+	if err != nil {
+		return errors.WithStackIf(err)
+	}
+	_ = resp.Body.Close()
+	return nil
+}
+
 // getServersPaged returns a subset of servers from the Panel API using the
 // pagination query parameters.
 func (c *client) getServersPaged(ctx context.Context, page, limit int) ([]RawServerData, Pagination, error) {

remote/types.go

@@ -2,15 +2,19 @@ package remote
 
 import (
 	"bytes"
+	"github.com/apex/log"
+	"github.com/goccy/go-json"
 	"regexp"
 	"strings"
 
-	"github.com/apex/log"
-	"github.com/goccy/go-json"
-
 	"github.com/pterodactyl/wings/parser"
 )
 
+const (
+	SftpAuthPassword  = SftpAuthRequestType("password")
+	SftpAuthPublicKey = SftpAuthRequestType("public_key")
+)
+
 // A generic type allowing for easy binding use when making requests to API
 // endpoints that only expect a singular argument or something that would not
 // benefit from being a typed struct.
@@ -63,14 +67,17 @@ type RawServerData struct {
 	ProcessConfiguration json.RawMessage `json:"process_configuration"`
 }
 
+type SftpAuthRequestType string
+
 // SftpAuthRequest defines the request details that are passed along to the Panel
 // when determining if the credentials provided to Wings are valid.
 type SftpAuthRequest struct {
-	User          string `json:"username"`
-	Pass          string `json:"password"`
-	IP            string `json:"ip"`
-	SessionID     []byte `json:"session_id"`
-	ClientVersion []byte `json:"client_version"`
+	Type          SftpAuthRequestType `json:"type"`
+	User          string              `json:"username"`
+	Pass          string              `json:"password"`
+	IP            string              `json:"ip"`
+	SessionID     []byte              `json:"session_id"`
+	ClientVersion []byte              `json:"client_version"`
 }
 
 // SftpAuthResponse is returned by the Panel when a pair of SFTP credentials
@@ -79,7 +86,6 @@ type SftpAuthRequest struct {
 // user for the SFTP subsystem.
 type SftpAuthResponse struct {
 	Server      string   `json:"server"`
-	Token       string   `json:"token"`
 	Permissions []string `json:"permissions"`
 }
 

router/downloader/downloader.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"mime"
 	"net"
 	"net/http"
 	"net/url"
@@ -13,8 +14,8 @@ import (
 	"time"
 
 	"emperror.dev/errors"
-	"github.com/google/uuid"
 	"github.com/goccy/go-json"
+	"github.com/google/uuid"
 
 	"github.com/pterodactyl/wings/server"
 )
@@ -77,10 +78,13 @@ func (c *Counter) Write(p []byte) (int, error) {
 type DownloadRequest struct {
 	Directory string
 	URL       *url.URL
+	FileName  string
+	UseHeader bool
 }
 
 type Download struct {
 	Identifier string
+	path       string
 	mu         sync.RWMutex
 	req        DownloadRequest
 	server     *server.Server
@@ -172,8 +176,28 @@ func (dl *Download) Execute() error {
 		}
 	}
 
-	fnameparts := strings.Split(dl.req.URL.Path, "/")
-	p := filepath.Join(dl.req.Directory, fnameparts[len(fnameparts)-1])
+	if dl.req.UseHeader {
+		if contentDisposition := res.Header.Get("Content-Disposition"); contentDisposition != "" {
+			_, params, err := mime.ParseMediaType(contentDisposition)
+			if err != nil {
+				return errors.WrapIf(err, "downloader: invalid \"Content-Disposition\" header")
+			}
+
+			if v, ok := params["filename"]; ok {
+				dl.path = v
+			}
+		}
+	}
+	if dl.path == "" {
+		if dl.req.FileName != "" {
+			dl.path = dl.req.FileName
+		} else {
+			parts := strings.Split(dl.req.URL.Path, "/")
+			dl.path = parts[len(parts)-1]
+		}
+	}
+
+	p := dl.Path()
 	dl.server.Log().WithField("path", p).Debug("writing remote file to disk")
 
 	r := io.TeeReader(res.Body, dl.counter(res.ContentLength))
@@ -205,6 +229,10 @@ func (dl *Download) Progress() float64 {
 	return dl.progress
 }
 
+func (dl *Download) Path() string {
+	return filepath.Join(dl.req.Directory, dl.path)
+}
+
 // Handles a write event by updating the progress completed percentage and firing off
 // events to the server websocket as needed.
 func (dl *Download) counter(contentLength int64) *Counter {

router/router.go

@@ -66,6 +66,7 @@ func Configure(m *wserver.Manager, client remote.Client) *gin.Engine {
 		server.DELETE("", deleteServer)
 
 		server.GET("/logs", getServerLogs)
+		server.GET("/activity", getServerActivityLogs)
 		server.POST("/power", postServerPower)
 		server.POST("/commands", postServerCommands)
 		server.POST("/install", postServerInstall)

router/router_server.go

@@ -2,6 +2,9 @@ package router
 
 import (
 	"context"
+	"github.com/goccy/go-json"
+	"github.com/pterodactyl/wings/internal/database"
+	"github.com/xujiajun/nutsdb"
 	"net/http"
 	"os"
 	"strconv"
@@ -40,6 +43,44 @@ func getServerLogs(c *gin.Context) {
 	c.JSON(http.StatusOK, gin.H{"data": out})
 }
 
+// Returns the activity logs tracked internally for the server instance. Note that these
+// logs are routinely cleared out as Wings communicates directly with the Panel to pass
+// along all of the logs for servers it monitors. As activities are passed to the panel
+// they are deleted from Wings.
+//
+// As a result, this endpoint may or may not return data, and the data returned can change
+// between requests.
+func getServerActivityLogs(c *gin.Context) {
+	s := ExtractServer(c)
+
+	var out [][]byte
+	err := database.DB().View(func(tx *nutsdb.Tx) error {
+		items, err := tx.LRange(database.ServerActivityBucket, []byte(s.ID()), 0, 10)
+		if err != nil {
+			return err
+		}
+		out = items
+		return nil
+	})
+
+	if err != nil {
+		middleware.CaptureAndAbort(c, err)
+		return
+	}
+
+	var activity []*server.Activity
+	for _, b := range out {
+		var a server.Activity
+		if err := json.Unmarshal(b, &a); err != nil {
+			middleware.CaptureAndAbort(c, err)
+			return
+		}
+		activity = append(activity, &a)
+	}
+
+	c.JSON(http.StatusOK, gin.H{"data": activity})
+}
+
 // Handles a request to control the power state of a server. If the action being passed
 // through is invalid a 404 is returned. Otherwise, a HTTP/202 Accepted response is returned
 // and the actual power action is run asynchronously so that we don't have to block the

router/router_server_files.go

@@ -13,6 +13,8 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/pterodactyl/wings/config"
+
 	"emperror.dev/errors"
 	"github.com/apex/log"
 	"github.com/gin-gonic/gin"
@@ -35,6 +37,15 @@ func getServerFileContents(c *gin.Context) {
 		return
 	}
 	defer f.Close()
+	// Don't allow a named pipe to be opened.
+	//
+	// @see https://github.com/pterodactyl/panel/issues/4059
+	if st.Mode()&os.ModeNamedPipe != 0 {
+		c.AbortWithStatusJSON(http.StatusBadRequest, gin.H{
+			"error": "Cannot open files of this type.",
+		})
+		return
+	}
 
 	c.Header("X-Mime-Type", st.Mimetype)
 	c.Header("Content-Length", strconv.Itoa(int(st.Size())))
@@ -120,6 +131,10 @@ func putServerRenameFiles(c *gin.Context) {
 					// Return nil if the error is an is not exists.
 					// NOTE: os.IsNotExist() does not work if the error is wrapped.
 					if errors.Is(err, os.ErrNotExist) {
+						s.Log().WithField("error", err).
+							WithField("from_path", pf).
+							WithField("to_path", pt).
+							Warn("failed to rename: source or target does not exist")
 						return nil
 					}
 					return err
@@ -255,9 +270,12 @@ func postServerPullRemoteFile(c *gin.Context) {
 	s := ExtractServer(c)
 	var data struct {
 		// Deprecated
-		Directory string `binding:"required_without=RootPath,omitempty" json:"directory"`
-		RootPath  string `binding:"required_without=Directory,omitempty" json:"root"`
-		URL       string `binding:"required" json:"url"`
+		Directory  string `binding:"required_without=RootPath,omitempty" json:"directory"`
+		RootPath   string `binding:"required_without=Directory,omitempty" json:"root"`
+		URL        string `binding:"required" json:"url"`
+		FileName   string `json:"file_name"`
+		UseHeader  bool   `json:"use_header"`
+		Foreground bool   `json:"foreground"`
 	}
 	if err := c.BindJSON(&data); err != nil {
 		return
@@ -295,21 +313,41 @@ func postServerPullRemoteFile(c *gin.Context) {
 	dl := downloader.New(s, downloader.DownloadRequest{
 		Directory: data.RootPath,
 		URL:       u,
+		FileName:  data.FileName,
+		UseHeader: data.UseHeader,
 	})
 
-	// Execute this pull in a separate thread since it may take a long time to complete.
-	go func() {
+	download := func() error {
 		s.Log().WithField("download_id", dl.Identifier).WithField("url", u.String()).Info("starting pull of remote file to disk")
 		if err := dl.Execute(); err != nil {
 			s.Log().WithField("download_id", dl.Identifier).WithField("error", err).Error("failed to pull remote file")
+			return err
 		} else {
 			s.Log().WithField("download_id", dl.Identifier).Info("completed pull of remote file")
 		}
-	}()
+		return nil
+	}
+	if !data.Foreground {
+		go func() {
+			_ = download()
+		}()
+		c.JSON(http.StatusAccepted, gin.H{
+			"identifier": dl.Identifier,
+		})
+		return
+	}
 
-	c.JSON(http.StatusAccepted, gin.H{
-		"identifier": dl.Identifier,
-	})
+	if err := download(); err != nil {
+		NewServerError(err, s).Abort(c)
+		return
+	}
+
+	st, err := s.Filesystem().Stat(dl.Path())
+	if err != nil {
+		NewServerError(err, s).AbortFilesystemError(c)
+		return
+	}
+	c.JSON(http.StatusOK, &st)
 }
 
 // Stops a remote file download if it exists and belongs to this server.
@@ -537,8 +575,16 @@ func postServerUploadFiles(c *gin.Context) {
 
 	directory := c.Query("directory")
 
+	maxFileSize := config.Get().Api.UploadLimit
+	maxFileSizeBytes := maxFileSize * 1024 * 1024
 	var totalSize int64
 	for _, header := range headers {
+		if header.Size > maxFileSizeBytes {
+			c.AbortWithStatusJSON(http.StatusBadRequest, gin.H{
+				"error": "File " + header.Filename + " is larger than the maximum file upload size of " + strconv.FormatInt(maxFileSize, 10) + " MB.",
+			})
+			return
+		}
 		totalSize += header.Size
 	}
 

router/router_server_ws.go

@@ -5,8 +5,8 @@ import (
 	"time"
 
 	"github.com/gin-gonic/gin"
-	ws "github.com/gorilla/websocket"
 	"github.com/goccy/go-json"
+	ws "github.com/gorilla/websocket"
 
 	"github.com/pterodactyl/wings/router/middleware"
 	"github.com/pterodactyl/wings/router/websocket"

router/router_transfer.go

@@ -178,7 +178,7 @@ func postServerArchive(c *gin.Context) {
 
 		// Ensure the server is offline. Sometimes a "No such container" error gets through
 		// which means the server is already stopped. We can ignore that.
-		if err := s.Environment.WaitForStop(60, false); err != nil && !strings.Contains(strings.ToLower(err.Error()), "no such container") {
+		if err := s.Environment.WaitForStop(s.Context(), time.Minute, false); err != nil && !strings.Contains(strings.ToLower(err.Error()), "no such container") {
 			sendTransferLog("Failed to stop server, aborting transfer..")
 			l.WithField("error", err).Error("failed to stop server")
 			return

router/tokens/websocket.go

@@ -7,7 +7,6 @@ import (
 
 	"github.com/apex/log"
 	"github.com/gbrlsnchs/jwt/v3"
-	"github.com/goccy/go-json"
 )
 
 // The time at which Wings was booted. No JWT's created before this time are allowed to
@@ -35,15 +34,15 @@ func DenyJTI(jti string) {
 	denylist.Store(jti, time.Now())
 }
 
-// A JWT payload for Websocket connections. This JWT is passed along to the Websocket after
-// it has been connected to by sending an "auth" event.
+// WebsocketPayload defines the JWT payload for a websocket connection. This JWT is passed along to
+// the websocket after it has been connected to by sending an "auth" event.
 type WebsocketPayload struct {
 	jwt.Payload
 	sync.RWMutex
 
-	UserID      json.Number `json:"user_id"`
-	ServerUUID  string      `json:"server_uuid"`
-	Permissions []string    `json:"permissions"`
+	UserUUID    string   `json:"user_uuid"`
+	ServerUUID  string   `json:"server_uuid"`
+	Permissions []string `json:"permissions"`
 }
 
 // Returns the JWT payload.

router/websocket/listeners.go

@@ -7,8 +7,9 @@ import (
 
 	"emperror.dev/errors"
 	"github.com/goccy/go-json"
-
 	"github.com/pterodactyl/wings/events"
+	"github.com/pterodactyl/wings/system"
+
 	"github.com/pterodactyl/wings/server"
 )
 
@@ -88,12 +89,13 @@ func (h *Handler) listenForServerEvents(ctx context.Context) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	eventChan := make(chan events.Event)
+	eventChan := make(chan []byte)
 	logOutput := make(chan []byte, 8)
 	installOutput := make(chan []byte, 4)
-	h.server.Events().On(eventChan, e...)
-	h.server.Sink(server.LogSink).On(logOutput)
-	h.server.Sink(server.InstallSink).On(installOutput)
+
+	h.server.Events().On(eventChan) // TODO: make a sinky
+	h.server.Sink(system.LogSink).On(logOutput)
+	h.server.Sink(system.InstallSink).On(installOutput)
 
 	onError := func(evt string, err2 error) {
 		h.Logger().WithField("event", evt).WithField("error", err2).Error("failed to send event over server websocket")
@@ -110,19 +112,23 @@ func (h *Handler) listenForServerEvents(ctx context.Context) error {
 		select {
 		case <-ctx.Done():
 			break
-		case e := <-logOutput:
-			sendErr := h.SendJson(Message{Event: server.ConsoleOutputEvent, Args: []string{string(e)}})
+		case b := <-logOutput:
+			sendErr := h.SendJson(Message{Event: server.ConsoleOutputEvent, Args: []string{string(b)}})
 			if sendErr == nil {
 				continue
 			}
 			onError(server.ConsoleOutputEvent, sendErr)
-		case e := <-installOutput:
-			sendErr := h.SendJson(Message{Event: server.InstallOutputEvent, Args: []string{string(e)}})
+		case b := <-installOutput:
+			sendErr := h.SendJson(Message{Event: server.InstallOutputEvent, Args: []string{string(b)}})
 			if sendErr == nil {
 				continue
 			}
 			onError(server.InstallOutputEvent, sendErr)
-		case e := <-eventChan:
+		case b := <-eventChan:
+			var e events.Event
+			if err := events.DecodeTo(b, &e); err != nil {
+				continue
+			}
 			var sendErr error
 			message := Message{Event: e.Topic}
 			if str, ok := e.Data.(string); ok {
@@ -148,9 +154,9 @@ func (h *Handler) listenForServerEvents(ctx context.Context) error {
 	}
 
 	// These functions will automatically close the channel if it hasn't been already.
-	h.server.Events().Off(eventChan, e...)
-	h.server.Sink(server.LogSink).Off(logOutput)
-	h.server.Sink(server.InstallSink).Off(installOutput)
+	h.server.Events().Off(eventChan)
+	h.server.Sink(system.LogSink).Off(logOutput)
+	h.server.Sink(system.InstallSink).Off(installOutput)
 
 	// If the internal context is stopped it is either because the parent context
 	// got canceled or because we ran into an error. If the "err" variable is nil

router/websocket/websocket.go

@@ -11,9 +11,10 @@ import (
 	"emperror.dev/errors"
 	"github.com/apex/log"
 	"github.com/gbrlsnchs/jwt/v3"
+	"github.com/goccy/go-json"
 	"github.com/google/uuid"
 	"github.com/gorilla/websocket"
-	"github.com/goccy/go-json"
+	"github.com/pterodactyl/wings/system"
 
 	"github.com/pterodactyl/wings/config"
 	"github.com/pterodactyl/wings/environment"
@@ -39,6 +40,7 @@ type Handler struct {
 	Connection   *websocket.Conn `json:"-"`
 	jwt          *tokens.WebsocketPayload
 	server       *server.Server
+	ra           server.RequestActivity
 	uuid         uuid.UUID
 }
 
@@ -108,6 +110,7 @@ func GetHandler(s *server.Server, w http.ResponseWriter, r *http.Request) (*Hand
 		Connection: conn,
 		jwt:        nil,
 		server:     s,
+		ra:         s.NewRequestActivity("", r.RemoteAddr),
 		uuid:       u,
 	}, nil
 }
@@ -263,6 +266,7 @@ func (h *Handler) GetJwt() *tokens.WebsocketPayload {
 // setJwt sets the JWT for the websocket in a race-safe manner.
 func (h *Handler) setJwt(token *tokens.WebsocketPayload) {
 	h.Lock()
+	h.ra = h.ra.SetUser(token.UserUUID)
 	h.jwt = token
 	h.Unlock()
 }
@@ -353,7 +357,7 @@ func (h *Handler) HandleInbound(ctx context.Context, m Message) error {
 			}
 
 			err := h.server.HandlePowerAction(action)
-			if errors.Is(err, context.DeadlineExceeded) {
+			if errors.Is(err, system.ErrLockerLocked) {
 				m, _ := h.GetErrorMessage("another power action is currently being processed for this server, please try again later")
 
 				_ = h.SendJson(Message{
@@ -364,6 +368,10 @@ func (h *Handler) HandleInbound(ctx context.Context, m Message) error {
 				return nil
 			}
 
+			if err == nil {
+				_ = h.ra.Save(h.server, server.Event(server.ActivityPowerPrefix+action), nil)
+			}
+
 			return err
 		}
 	case SendServerLogsEvent:
@@ -420,6 +428,10 @@ func (h *Handler) HandleInbound(ctx context.Context, m Message) error {
 				}
 			}
 
+			_ = h.ra.Save(h.server, server.ActivityConsoleCommand, server.ActivityMeta{
+				"command": strings.Join(m.Args, ""),
+			})
+
 			return h.server.Environment.SendCommand(strings.Join(m.Args, ""))
 		}
 	}

server/activity.go

@@ -0,0 +1,130 @@
+package server
+
+import (
+	"emperror.dev/errors"
+	"github.com/apex/log"
+	"github.com/goccy/go-json"
+	"github.com/pterodactyl/wings/internal/database"
+	"github.com/xujiajun/nutsdb"
+	"regexp"
+	"time"
+)
+
+type Event string
+type ActivityMeta map[string]interface{}
+
+const ActivityPowerPrefix = "power_"
+
+const (
+	ActivityConsoleCommand = Event("console_command")
+)
+
+var ipTrimRegex = regexp.MustCompile(`(:\d*)?$`)
+
+type Activity struct {
+	// User is UUID of the user that triggered this event, or an empty string if the event
+	// cannot be tied to a specific user, in which case we will assume it was the system
+	// user.
+	User string `json:"user"`
+	// Server is the UUID of the server this event is associated with.
+	Server string `json:"server"`
+	// Event is a string that describes what occurred, and is used by the Panel instance to
+	// properly associate this event in the activity logs.
+	Event Event `json:"event"`
+	// Metadata is either a null value, string, or a JSON blob with additional event specific
+	// metadata that can be provided.
+	Metadata ActivityMeta `json:"metadata"`
+	// IP is the IP address that triggered this event, or an empty string if it cannot be
+	// determined properly.
+	IP        string    `json:"ip"`
+	Timestamp time.Time `json:"timestamp"`
+}
+
+// RequestActivity is a wrapper around a LoggedEvent that is able to track additional request
+// specific metadata including the specific user and IP address associated with all subsequent
+// events. The internal logged event structure can be extracted by calling RequestEvent.Event().
+type RequestActivity struct {
+	server string
+	user   string
+	ip     string
+}
+
+// Event returns the underlying logged event from the RequestEvent instance and sets the
+// specific event and metadata on it.
+func (ra RequestActivity) Event(event Event, metadata ActivityMeta) Activity {
+	return Activity{
+		User:     ra.user,
+		Server:   ra.server,
+		IP:       ra.ip,
+		Event:    event,
+		Metadata: metadata,
+	}
+}
+
+// Save creates a new event instance and saves it. If an error is encountered it is automatically
+// logged to the provided server's error logging output. The error is also returned to the caller
+// but can be ignored.
+func (ra RequestActivity) Save(s *Server, event Event, metadata ActivityMeta) error {
+	if err := ra.Event(event, metadata).Save(); err != nil {
+		s.Log().WithField("error", err).WithField("event", event).Error("activity: failed to save event")
+		return errors.WithStack(err)
+	}
+	return nil
+}
+
+// IP returns the IP address associated with this entry.
+func (ra RequestActivity) IP() string {
+	return ra.ip
+}
+
+// SetUser clones the RequestActivity struct and sets a new user value on the copy
+// before returning it.
+func (ra RequestActivity) SetUser(u string) RequestActivity {
+	c := ra
+	c.user = u
+	return c
+}
+
+// Save logs the provided event using Wings' internal K/V store so that we can then
+// pass it along to the Panel at set intervals. In addition, this will ensure that the events
+// are persisted to the disk, even between instance restarts.
+func (a Activity) Save() error {
+	if a.Timestamp.IsZero() {
+		a.Timestamp = time.Now().UTC()
+	}
+
+	// Since the "RemoteAddr" field can often include a port on the end we need to
+	// trim that off, otherwise it'll fail validation when sent to the Panel.
+	a.IP = ipTrimRegex.ReplaceAllString(a.IP, "")
+
+	value, err := json.Marshal(a)
+	if err != nil {
+		return errors.Wrap(err, "database: failed to marshal activity into json bytes")
+	}
+
+	return database.DB().Update(func(tx *nutsdb.Tx) error {
+		log.WithField("subsystem", "activity").
+			WithFields(log.Fields{"server": a.Server, "user": a.User, "event": a.Event, "ip": a.IP}).
+			Debug("saving activity to database")
+
+		if err := tx.RPush(database.ServerActivityBucket, []byte("events"), value); err != nil {
+			return errors.WithStack(err)
+		}
+		return nil
+	})
+}
+
+func (s *Server) NewRequestActivity(user string, ip string) RequestActivity {
+	return RequestActivity{server: s.ID(), user: user, ip: ip}
+}
+
+// NewActivity creates a new event instance for the server in question.
+func (s *Server) NewActivity(user string, event Event, metadata ActivityMeta, ip string) Activity {
+	return Activity{
+		User:     user,
+		Server:   s.ID(),
+		Event:    event,
+		Metadata: metadata,
+		IP:       ip,
+	}
+}

server/backup.go

@@ -142,7 +142,7 @@ func (s *Server) RestoreBackup(b backup.BackupInterface, reader io.ReadCloser) (
 	// instance, otherwise you'll likely hit all types of write errors due to the
 	// server being suspended.
 	if s.Environment.State() != environment.ProcessOfflineState {
-		if err = s.Environment.WaitForStop(120, false); err != nil {
+		if err = s.Environment.WaitForStop(s.Context(), time.Minute*2, false); err != nil {
 			if !client.IsErrNotFound(err) {
 				return errors.WrapIf(err, "server/backup: restore: failed to wait for container stop")
 			}

server/config_parser.go

@@ -6,12 +6,14 @@ import (
 	"github.com/gammazero/workerpool"
 )
 
-// Parent function that will update all of the defined configuration files for a server
-// automatically to ensure that they always use the specified values.
+// UpdateConfigurationFiles updates all of the defined configuration files for
+// a server automatically to ensure that they always use the specified values.
 func (s *Server) UpdateConfigurationFiles() {
 	pool := workerpool.New(runtime.NumCPU())
 
+	s.Log().Debug("acquiring process configuration files...")
 	files := s.ProcessConfiguration().ConfigurationFiles
+	s.Log().Debug("acquired process configuration files")
 	for _, cf := range files {
 		f := cf
 
@@ -26,6 +28,8 @@ func (s *Server) UpdateConfigurationFiles() {
 			if err := f.Parse(p, false); err != nil {
 				s.Log().WithField("error", err).Error("failed to parse and update server configuration file")
 			}
+
+			s.Log().WithField("path", f.FileName).Debug("finished processing server configuration file")
 		})
 	}
 

server/console_test.go

@@ -19,7 +19,7 @@ func TestName(t *testing.T) {
 		})
 
 		g.It("calls strike once per time period", func() {
-			t := newConsoleThrottle(1, time.Millisecond * 20)
+			t := newConsoleThrottle(1, time.Millisecond*20)
 
 			var times int
 			t.strike = func() {
@@ -53,10 +53,10 @@ func TestName(t *testing.T) {
 }
 
 func BenchmarkConsoleThrottle(b *testing.B) {
-	t := newConsoleThrottle(10, time.Millisecond * 10)
+	t := newConsoleThrottle(10, time.Millisecond*10)
 
 	b.ReportAllocs()
-    for i := 0; i < b.N; i++ {
-        t.Allow()
-    }
+	for i := 0; i < b.N; i++ {
+		t.Allow()
+	}
 }

server/events.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"github.com/pterodactyl/wings/events"
+	"github.com/pterodactyl/wings/system"
 )
 
 // Defines all of the possible output events for a server.
@@ -20,7 +21,7 @@ const (
 	TransferStatusEvent         = "transfer status"
 )
 
-// Returns the server's emitter instance.
+// Events returns the server's emitter instance.
 func (s *Server) Events() *events.Bus {
 	s.emitterLock.Lock()
 	defer s.emitterLock.Unlock()
@@ -31,3 +32,24 @@ func (s *Server) Events() *events.Bus {
 
 	return s.emitter
 }
+
+// Sink returns the instantiated and named sink for a server. If the sink has
+// not been configured yet this function will cause a panic condition.
+func (s *Server) Sink(name system.SinkName) *system.SinkPool {
+	sink, ok := s.sinks[name]
+	if !ok {
+		s.Log().Fatalf("attempt to access nil sink: %s", name)
+	}
+	return sink
+}
+
+// DestroyAllSinks iterates over all of the sinks configured for the server and
+// destroys their instances. Note that this will cause a panic if you attempt
+// to call Server.Sink() again after. This function is only used when a server
+// is being deleted from the system.
+func (s *Server) DestroyAllSinks() {
+	s.Log().Info("destroying all registered sinks for server instance")
+	for _, sink := range s.sinks {
+		sink.Destroy()
+	}
+}

server/filesystem/archive.go

@@ -130,7 +130,7 @@ func (a *Archive) withFilesCallback(tw *tar.Writer) func(path string, de *godirw
 		for _, f := range a.Files {
 			// If the given doesn't match, or doesn't have the same prefix continue
 			// to the next item in the loop.
-			if p != f && !strings.HasPrefix(p, f) {
+			if p != f && !strings.HasPrefix(strings.TrimSuffix(p, "/")+"/", f) {
 				continue
 			}
 

server/filesystem/compress.go

@@ -5,9 +5,12 @@ import (
 	"archive/zip"
 	"compress/gzip"
 	"fmt"
+	gzip2 "github.com/klauspost/compress/gzip"
+	zip2 "github.com/klauspost/compress/zip"
 	"os"
 	"path"
 	"path/filepath"
+	"reflect"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -172,13 +175,26 @@ func ExtractNameFromArchive(f archiver.File) string {
 		return f.Name()
 	}
 	switch s := sys.(type) {
+	case *zip.FileHeader:
+		return s.Name
+	case *zip2.FileHeader:
+		return s.Name
 	case *tar.Header:
 		return s.Name
 	case *gzip.Header:
 		return s.Name
-	case *zip.FileHeader:
+	case *gzip2.Header:
 		return s.Name
 	default:
+		// At this point we cannot figure out what type of archive this might be so
+		// just try to find the name field in the struct. If it is found return it.
+		field := reflect.Indirect(reflect.ValueOf(sys)).FieldByName("Name")
+		if field.IsValid() {
+			return field.String()
+		}
+		// Fallback to the basename of the file at this point. There is nothing we can really
+		// do to try and figure out what the underlying directory of the file is supposed to
+		// be since it didn't implement a name field.
 		return f.Name()
 	}
 }

server/filesystem/filesystem.go

@@ -115,19 +115,6 @@ func (fs *Filesystem) Touch(p string, flag int) (*os.File, error) {
 	return f, nil
 }
 
-// Reads a file on the system and returns it as a byte representation in a file
-// reader. This is not the most memory efficient usage since it will be reading the
-// entirety of the file into memory.
-func (fs *Filesystem) Readfile(p string, w io.Writer) error {
-	file, _, err := fs.File(p)
-	if err != nil {
-		return err
-	}
-	defer file.Close()
-	_, err = bufio.NewReader(file).WriteTo(w)
-	return err
-}
-
 // Writefile writes a file to the system. If the file does not already exist one
 // will be created. This will also properly recalculate the disk space used by
 // the server when writing new files or modifying existing ones.
@@ -184,16 +171,16 @@ func (fs *Filesystem) CreateDirectory(name string, p string) error {
 	return os.MkdirAll(cleaned, 0o755)
 }
 
-// Moves (or renames) a file or directory.
+// Rename moves (or renames) a file or directory.
 func (fs *Filesystem) Rename(from string, to string) error {
 	cleanedFrom, err := fs.SafePath(from)
 	if err != nil {
-		return err
+		return errors.WithStack(err)
 	}
 
 	cleanedTo, err := fs.SafePath(to)
 	if err != nil {
-		return err
+		return errors.WithStack(err)
 	}
 
 	// If the target file or directory already exists the rename function will fail, so just
@@ -215,7 +202,10 @@ func (fs *Filesystem) Rename(from string, to string) error {
 		}
 	}
 
-	return os.Rename(cleanedFrom, cleanedTo)
+	if err := os.Rename(cleanedFrom, cleanedTo); err != nil {
+		return errors.WithStack(err)
+	}
+	return nil
 }
 
 // Recursively iterates over a file or directory and sets the permissions on all of the
@@ -492,7 +482,11 @@ func (fs *Filesystem) ListDirectory(p string) ([]Stat, error) {
 					cleanedp, _ = fs.SafePath(filepath.Join(cleaned, f.Name()))
 				}
 
-				if cleanedp != "" {
+				// Don't try to detect the type on a pipe — this will just hang the application and
+				// you'll never get a response back.
+				//
+				// @see https://github.com/pterodactyl/panel/issues/4059
+				if cleanedp != "" && f.Mode()&os.ModeNamedPipe == 0 {
 					m, _ = mimetype.DetectFile(filepath.Join(cleaned, f.Name()))
 				} else {
 					// Just pass this for an unknown type because the file could not safely be resolved within

server/filesystem/filesystem_test.go

@@ -1,6 +1,7 @@
 package filesystem
 
 import (
+	"bufio"
 	"bytes"
 	"errors"
 	"math/rand"
@@ -44,6 +45,14 @@ type rootFs struct {
 	root string
 }
 
+func getFileContent(file *os.File) string {
+	var w bytes.Buffer
+	if _, err := bufio.NewReader(file).WriteTo(&w); err != nil {
+		panic(err)
+	}
+	return w.String()
+}
+
 func (rfs *rootFs) CreateServerFile(p string, c []byte) error {
 	f, err := os.Create(filepath.Join(rfs.root, "/server", p))
 
@@ -75,54 +84,6 @@ func (rfs *rootFs) reset() {
 	}
 }
 
-func TestFilesystem_Readfile(t *testing.T) {
-	g := Goblin(t)
-	fs, rfs := NewFs()
-
-	g.Describe("Readfile", func() {
-		buf := &bytes.Buffer{}
-
-		g.It("opens a file if it exists on the system", func() {
-			err := rfs.CreateServerFileFromString("test.txt", "testing")
-			g.Assert(err).IsNil()
-
-			err = fs.Readfile("test.txt", buf)
-			g.Assert(err).IsNil()
-			g.Assert(buf.String()).Equal("testing")
-		})
-
-		g.It("returns an error if the file does not exist", func() {
-			err := fs.Readfile("test.txt", buf)
-			g.Assert(err).IsNotNil()
-			g.Assert(errors.Is(err, os.ErrNotExist)).IsTrue()
-		})
-
-		g.It("returns an error if the \"file\" is a directory", func() {
-			err := os.Mkdir(filepath.Join(rfs.root, "/server/test.txt"), 0o755)
-			g.Assert(err).IsNil()
-
-			err = fs.Readfile("test.txt", buf)
-			g.Assert(err).IsNotNil()
-			g.Assert(IsErrorCode(err, ErrCodeIsDirectory)).IsTrue()
-		})
-
-		g.It("cannot open a file outside the root directory", func() {
-			err := rfs.CreateServerFileFromString("/../test.txt", "testing")
-			g.Assert(err).IsNil()
-
-			err = fs.Readfile("/../test.txt", buf)
-			g.Assert(err).IsNotNil()
-			g.Assert(IsErrorCode(err, ErrCodePathResolution)).IsTrue()
-		})
-
-		g.AfterEach(func() {
-			buf.Truncate(0)
-			atomic.StoreInt64(&fs.diskUsed, 0)
-			rfs.reset()
-		})
-	})
-}
-
 func TestFilesystem_Writefile(t *testing.T) {
 	g := Goblin(t)
 	fs, rfs := NewFs()
@@ -140,9 +101,10 @@ func TestFilesystem_Writefile(t *testing.T) {
 			err := fs.Writefile("test.txt", r)
 			g.Assert(err).IsNil()
 
-			err = fs.Readfile("test.txt", buf)
+			f, _, err := fs.File("test.txt")
 			g.Assert(err).IsNil()
-			g.Assert(buf.String()).Equal("test file content")
+			defer f.Close()
+			g.Assert(getFileContent(f)).Equal("test file content")
 			g.Assert(atomic.LoadInt64(&fs.diskUsed)).Equal(r.Size())
 		})
 
@@ -152,9 +114,10 @@ func TestFilesystem_Writefile(t *testing.T) {
 			err := fs.Writefile("/some/nested/test.txt", r)
 			g.Assert(err).IsNil()
 
-			err = fs.Readfile("/some/nested/test.txt", buf)
+			f, _, err := fs.File("/some/nested/test.txt")
 			g.Assert(err).IsNil()
-			g.Assert(buf.String()).Equal("test file content")
+			defer f.Close()
+			g.Assert(getFileContent(f)).Equal("test file content")
 		})
 
 		g.It("can create a new file inside a nested directory without a trailing slash", func() {
@@ -163,9 +126,10 @@ func TestFilesystem_Writefile(t *testing.T) {
 			err := fs.Writefile("some/../foo/bar/test.txt", r)
 			g.Assert(err).IsNil()
 
-			err = fs.Readfile("foo/bar/test.txt", buf)
+			f, _, err := fs.File("foo/bar/test.txt")
 			g.Assert(err).IsNil()
-			g.Assert(buf.String()).Equal("test file content")
+			defer f.Close()
+			g.Assert(getFileContent(f)).Equal("test file content")
 		})
 
 		g.It("cannot create a file outside the root directory", func() {
@@ -190,28 +154,6 @@ func TestFilesystem_Writefile(t *testing.T) {
 			g.Assert(IsErrorCode(err, ErrCodeDiskSpace)).IsTrue()
 		})
 
-		/*g.It("updates the total space used when a file is appended to", func() {
-			atomic.StoreInt64(&fs.diskUsed, 100)
-
-			b := make([]byte, 100)
-			_, _ = rand.Read(b)
-
-			r := bytes.NewReader(b)
-			err := fs.Writefile("test.txt", r)
-			g.Assert(err).IsNil()
-			g.Assert(atomic.LoadInt64(&fs.diskUsed)).Equal(int64(200))
-
-			// If we write less data than already exists, we should expect the total
-			// disk used to be decremented.
-			b = make([]byte, 50)
-			_, _ = rand.Read(b)
-
-			r = bytes.NewReader(b)
-			err = fs.Writefile("test.txt", r)
-			g.Assert(err).IsNil()
-			g.Assert(atomic.LoadInt64(&fs.diskUsed)).Equal(int64(150))
-		})*/
-
 		g.It("truncates the file when writing new contents", func() {
 			r := bytes.NewReader([]byte("original data"))
 			err := fs.Writefile("test.txt", r)
@@ -221,9 +163,10 @@ func TestFilesystem_Writefile(t *testing.T) {
 			err = fs.Writefile("test.txt", r)
 			g.Assert(err).IsNil()
 
-			err = fs.Readfile("test.txt", buf)
+			f, _, err := fs.File("test.txt")
 			g.Assert(err).IsNil()
-			g.Assert(buf.String()).Equal("new data")
+			defer f.Close()
+			g.Assert(getFileContent(f)).Equal("new data")
 		})
 
 		g.AfterEach(func() {

server/filesystem/path_test.go

@@ -119,16 +119,6 @@ func TestFilesystem_Blocks_Symlinks(t *testing.T) {
 		panic(err)
 	}
 
-	g.Describe("Readfile", func() {
-		g.It("cannot read a file symlinked outside the root", func() {
-			b := bytes.Buffer{}
-
-			err := fs.Readfile("symlinked.txt", &b)
-			g.Assert(err).IsNotNil()
-			g.Assert(IsErrorCode(err, ErrCodePathResolution)).IsTrue()
-		})
-	})
-
 	g.Describe("Writefile", func() {
 		g.It("cannot write to a file symlinked outside the root", func() {
 			r := bytes.NewReader([]byte("testing"))

server/install.go

@@ -10,6 +10,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"time"
 
 	"emperror.dev/errors"
 	"github.com/apex/log"
@@ -17,23 +18,23 @@ import (
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/api/types/mount"
 	"github.com/docker/docker/client"
-
 	"github.com/pterodactyl/wings/config"
 	"github.com/pterodactyl/wings/environment"
 	"github.com/pterodactyl/wings/remote"
 	"github.com/pterodactyl/wings/system"
 )
 
-// Executes the installation stack for a server process. Bubbles any errors up to the calling
-// function which should handle contacting the panel to notify it of the server state.
+// Install executes the installation stack for a server process. Bubbles any
+// errors up to the calling function which should handle contacting the panel to
+// notify it of the server state.
 //
-// Pass true as the first argument in order to execute a server sync before the process to
-// ensure the latest information is used.
+// Pass true as the first argument in order to execute a server sync before the
+// process to ensure the latest information is used.
 func (s *Server) Install(sync bool) error {
 	if sync {
 		s.Log().Info("syncing server state with remote source before executing installation process")
 		if err := s.Sync(); err != nil {
-			return err
+			return errors.WrapIf(err, "install: failed to sync server state with Panel")
 		}
 	}
 
@@ -57,7 +58,7 @@ func (s *Server) Install(sync bool) error {
 		// error to this log entry. Otherwise ignore it in this log since whatever is calling
 		// this function should handle the error and will end up logging the same one.
 		if err == nil {
-			l.WithField("error", serr)
+			l.WithField("error", err)
 		}
 
 		l.Warn("failed to notify panel of server install state")
@@ -71,7 +72,7 @@ func (s *Server) Install(sync bool) error {
 	// the install is completed.
 	s.Events().Publish(InstallCompletedEvent, "")
 
-	return err
+	return errors.WithStackIf(err)
 }
 
 // Reinstalls a server's software by utilizing the install script for the server egg. This
@@ -79,8 +80,8 @@ func (s *Server) Install(sync bool) error {
 func (s *Server) Reinstall() error {
 	if s.Environment.State() != environment.ProcessOfflineState {
 		s.Log().Debug("waiting for server instance to enter a stopped state")
-		if err := s.Environment.WaitForStop(10, true); err != nil {
-			return err
+		if err := s.Environment.WaitForStop(s.Context(), time.Second*10, true); err != nil {
+			return errors.WrapIf(err, "install: failed to stop running environment")
 		}
 	}
 
@@ -110,9 +111,7 @@ func (s *Server) internalInstall() error {
 type InstallationProcess struct {
 	Server *Server
 	Script *remote.InstallationScript
-
-	client  *client.Client
-	context context.Context
+	client *client.Client
 }
 
 // Generates a new installation process struct that will be used to create containers,
@@ -127,7 +126,6 @@ func NewInstallationProcess(s *Server, script *remote.InstallationScript) (*Inst
 		return nil, err
 	} else {
 		proc.client = c
-		proc.context = s.Context()
 	}
 
 	return proc, nil
@@ -157,7 +155,7 @@ func (s *Server) SetRestoring(state bool) {
 
 // Removes the installer container for the server.
 func (ip *InstallationProcess) RemoveContainer() error {
-	err := ip.client.ContainerRemove(ip.context, ip.Server.ID()+"_installer", types.ContainerRemoveOptions{
+	err := ip.client.ContainerRemove(ip.Server.Context(), ip.Server.ID()+"_installer", types.ContainerRemoveOptions{
 		RemoveVolumes: true,
 		Force:         true,
 	})
@@ -167,11 +165,10 @@ func (ip *InstallationProcess) RemoveContainer() error {
 	return nil
 }
 
-// Runs the installation process, this is done as in a background thread. This will configure
-// the required environment, and then spin up the installation container.
-//
-// Once the container finishes installing the results will be stored in an installation
-// log in the server's configuration directory.
+// Run runs the installation process, this is done as in a background thread.
+// This will configure the required environment, and then spin up the
+// installation container. Once the container finishes installing the results
+// are stored in an installation log in the server's configuration directory.
 func (ip *InstallationProcess) Run() error {
 	ip.Server.Log().Debug("acquiring installation process lock")
 	if !ip.Server.installing.SwapIf(true) {
@@ -207,7 +204,7 @@ func (ip *InstallationProcess) Run() error {
 
 // Returns the location of the temporary data for the installation process.
 func (ip *InstallationProcess) tempDir() string {
-	return filepath.Join(os.TempDir(), "pterodactyl/", ip.Server.ID())
+	return filepath.Join(config.Get().System.TmpDirectory, ip.Server.ID())
 }
 
 // Writes the installation script to a temporary file on the host machine so that it
@@ -267,9 +264,9 @@ func (ip *InstallationProcess) pullInstallationImage() error {
 		imagePullOptions.RegistryAuth = b64
 	}
 
-	r, err := ip.client.ImagePull(context.Background(), ip.Script.ContainerImage, imagePullOptions)
+	r, err := ip.client.ImagePull(ip.Server.Context(), ip.Script.ContainerImage, imagePullOptions)
 	if err != nil {
-		images, ierr := ip.client.ImageList(context.Background(), types.ImageListOptions{})
+		images, ierr := ip.client.ImageList(ip.Server.Context(), types.ImageListOptions{})
 		if ierr != nil {
 			// Well damn, something has gone really wrong here, just go ahead and abort there
 			// isn't much anything we can do to try and self-recover from this.
@@ -312,9 +309,10 @@ func (ip *InstallationProcess) pullInstallationImage() error {
 	return nil
 }
 
-// Runs before the container is executed. This pulls down the required docker container image
-// as well as writes the installation script to the disk. This process is executed in an async
-// manner, if either one fails the error is returned.
+// BeforeExecute runs before the container is executed. This pulls down the
+// required docker container image as well as writes the installation script to
+// the disk. This process is executed in an async manner, if either one fails
+// the error is returned.
 func (ip *InstallationProcess) BeforeExecute() error {
 	if err := ip.writeScriptToDisk(); err != nil {
 		return errors.WithMessage(err, "failed to write installation script to disk")
@@ -340,7 +338,7 @@ func (ip *InstallationProcess) AfterExecute(containerId string) error {
 	defer ip.RemoveContainer()
 
 	ip.Server.Log().WithField("container_id", containerId).Debug("pulling installation logs for server")
-	reader, err := ip.client.ContainerLogs(ip.context, containerId, types.ContainerLogsOptions{
+	reader, err := ip.client.ContainerLogs(ip.Server.Context(), containerId, types.ContainerLogsOptions{
 		ShowStdout: true,
 		ShowStderr: true,
 		Follow:     false,
@@ -395,12 +393,13 @@ func (ip *InstallationProcess) AfterExecute(containerId string) error {
 	return nil
 }
 
-// Executes the installation process inside a specially created docker container.
+// Execute executes the installation process inside a specially created docker
+// container.
 func (ip *InstallationProcess) Execute() (string, error) {
 	// Create a child context that is canceled once this function is done running. This
 	// will also be canceled if the parent context (from the Server struct) is canceled
 	// which occurs if the server is deleted.
-	ctx, cancel := context.WithCancel(ip.context)
+	ctx, cancel := context.WithCancel(ip.Server.Context())
 	defer cancel()
 
 	conf := &container.Config{
@@ -511,18 +510,15 @@ func (ip *InstallationProcess) Execute() (string, error) {
 // the server configuration directory, as well as to a websocket listener so
 // that the process can be viewed in the panel by administrators.
 func (ip *InstallationProcess) StreamOutput(ctx context.Context, id string) error {
-	reader, err := ip.client.ContainerLogs(ctx, id, types.ContainerLogsOptions{
-		ShowStdout: true,
-		ShowStderr: true,
-		Follow:     true,
-	})
+	opts := types.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Follow: true}
+	reader, err := ip.client.ContainerLogs(ctx, id, opts)
 	if err != nil {
 		return err
 	}
 	defer reader.Close()
 
-	err = system.ScanReader(reader, ip.Server.Sink(InstallSink).Push)
-	if err != nil {
+	err = system.ScanReader(reader, ip.Server.Sink(system.InstallSink).Push)
+	if err != nil && !errors.Is(err, context.Canceled) {
 		ip.Server.Log().WithFields(log.Fields{"container_id": id, "error": err}).Warn("error processing install output lines")
 	}
 	return nil

server/listeners.go

@@ -5,11 +5,13 @@ import (
 	"regexp"
 	"strconv"
 	"sync"
+	"time"
 
 	"github.com/apex/log"
+	"github.com/pterodactyl/wings/events"
+	"github.com/pterodactyl/wings/system"
 
 	"github.com/pterodactyl/wings/environment"
-	"github.com/pterodactyl/wings/events"
 	"github.com/pterodactyl/wings/remote"
 )
 
@@ -44,7 +46,7 @@ func (dsl *diskSpaceLimiter) Reset() {
 func (dsl *diskSpaceLimiter) Trigger() {
 	dsl.o.Do(func() {
 		dsl.server.PublishConsoleOutputFromDaemon("Server is exceeding the assigned disk space limit, stopping process now.")
-		if err := dsl.server.Environment.WaitForStop(60, true); err != nil {
+		if err := dsl.server.Environment.WaitForStop(dsl.server.Context(), time.Minute, true); err != nil {
 			dsl.server.Log().WithField("error", err).Error("failed to stop server after exceeding space limit!")
 		}
 	})
@@ -72,47 +74,57 @@ func (s *Server) processConsoleOutputEvent(v []byte) {
 		return
 	}
 
-	s.Sink(LogSink).Push(v)
+	s.Sink(system.LogSink).Push(v)
 }
 
 // StartEventListeners adds all the internal event listeners we want to use for
 // a server. These listeners can only be removed by deleting the server as they
 // should last for the duration of the process' lifetime.
 func (s *Server) StartEventListeners() {
-	state := make(chan events.Event)
-	stats := make(chan events.Event)
-	docker := make(chan events.Event)
+	c := make(chan []byte, 8)
+	limit := newDiskLimiter(s)
+
+	s.Log().Debug("registering event listeners: console, state, resources...")
+	s.Environment.Events().On(c)
+	s.Environment.SetLogCallback(s.processConsoleOutputEvent)
 
 	go func() {
-		l := newDiskLimiter(s)
-
 		for {
 			select {
-			case e := <-state:
-				go func() {
-					// Reset the throttler when the process is started.
-					if e.Data == environment.ProcessStartingState {
-						l.Reset()
-						s.Throttler().Reset()
+			case v := <-c:
+				go func(v []byte, limit *diskSpaceLimiter) {
+					var e events.Event
+					if err := events.DecodeTo(v, &e); err != nil {
+						return
 					}
-
-					s.OnStateChange()
-				}()
-			case e := <-stats:
-				go func() {
-					s.resources.UpdateStats(e.Data.(environment.Stats))
-
-					// If there is no disk space available at this point, trigger the server
-					// disk limiter logic which will start to stop the running instance.
-					if !s.Filesystem().HasSpaceAvailable(true) {
-						l.Trigger()
-					}
-
-					s.Events().Publish(StatsEvent, s.Proc())
-				}()
-			case e := <-docker:
-				go func() {
 					switch e.Topic {
+					case environment.ResourceEvent:
+						{
+							var stats struct {
+								Topic string
+								Data  environment.Stats
+							}
+							if err := events.DecodeTo(v, &stats); err != nil {
+								s.Log().WithField("error", err).Warn("failed to decode server resource event")
+								return
+							}
+							s.resources.UpdateStats(stats.Data)
+							// If there is no disk space available at this point, trigger the server
+							// disk limiter logic which will start to stop the running instance.
+							if !s.Filesystem().HasSpaceAvailable(true) {
+								limit.Trigger()
+							}
+							s.Events().Publish(StatsEvent, s.Proc())
+						}
+					case environment.StateChangeEvent:
+						{
+							// Reset the throttler when the process is started.
+							if e.Data == environment.ProcessStartingState {
+								limit.Reset()
+								s.Throttler().Reset()
+							}
+							s.OnStateChange()
+						}
 					case environment.DockerImagePullStatus:
 						s.Events().Publish(InstallOutputEvent, e.Data)
 					case environment.DockerImagePullStarted:
@@ -120,18 +132,13 @@ func (s *Server) StartEventListeners() {
 					case environment.DockerImagePullCompleted:
 						s.PublishConsoleOutputFromDaemon("Finished pulling Docker container image")
 					default:
-						s.Log().WithField("topic", e.Topic).Error("unhandled docker event topic")
 					}
-				}()
+				}(v, limit)
+			case <-s.Context().Done():
+				return
 			}
 		}
 	}()
-
-	s.Log().Debug("registering event listeners: console, state, resources...")
-	s.Environment.SetLogCallback(s.processConsoleOutputEvent)
-	s.Environment.Events().On(state, environment.StateChangeEvent)
-	s.Environment.Events().On(stats, environment.ResourceEvent)
-	s.Environment.Events().On(docker, dockerEvents...)
 }
 
 var stripAnsiRegex = regexp.MustCompile("[\u001B\u009B][[\\]()#;?]*(?:(?:(?:[a-zA-Z\\d]*(?:;[a-zA-Z\\d]*)*)?\u0007)|(?:(?:\\d{1,4}(?:;\\d{0,4})*)?[\\dA-PRZcf-ntqry=><~]))")

server/power.go

@@ -133,11 +133,11 @@ func (s *Server) HandlePowerAction(action PowerAction, waitSeconds ...int) error
 
 		return s.Environment.Start(s.Context())
 	case PowerActionStop:
-		// We're specifically waiting for the process to be stopped here, otherwise the lock is released
-		// too soon, and you can rack up all sorts of issues.
-		return s.Environment.WaitForStop(10*60, true)
+		fallthrough
 	case PowerActionRestart:
-		if err := s.Environment.WaitForStop(10*60, true); err != nil {
+		// We're specifically waiting for the process to be stopped here, otherwise the lock is
+		// released too soon, and you can rack up all sorts of issues.
+		if err := s.Environment.WaitForStop(s.Context(), time.Minute*10, true); err != nil {
 			// Even timeout errors should be bubbled back up the stack. If the process didn't stop
 			// nicely, but the terminate argument was passed then the server is stopped without an
 			// error being returned.
@@ -149,6 +149,10 @@ func (s *Server) HandlePowerAction(action PowerAction, waitSeconds ...int) error
 			return err
 		}
 
+		if action == PowerActionStop {
+			return nil
+		}
+
 		// Now actually try to start the process by executing the normal pre-boot logic.
 		if err := s.onBeforeStart(); err != nil {
 			return err
@@ -156,7 +160,7 @@ func (s *Server) HandlePowerAction(action PowerAction, waitSeconds ...int) error
 
 		return s.Environment.Start(s.Context())
 	case PowerActionTerminate:
-		return s.Environment.Terminate(os.Kill)
+		return s.Environment.Terminate(s.Context(), os.Kill)
 	}
 
 	return errors.New("attempting to handle unknown power action")
@@ -197,15 +201,19 @@ func (s *Server) onBeforeStart() error {
 	// we don't need to actively do anything about it at this point, worse comes to worst the
 	// server starts in a weird state and the user can manually adjust.
 	s.PublishConsoleOutputFromDaemon("Updating process configuration files...")
+	s.Log().Debug("updating server configuration files...")
 	s.UpdateConfigurationFiles()
+	s.Log().Debug("updated server configuration files")
 
 	if config.Get().System.CheckPermissionsOnBoot {
 		s.PublishConsoleOutputFromDaemon("Ensuring file permissions are set correctly, this could take a few seconds...")
 		// Ensure all the server file permissions are set correctly before booting the process.
+		s.Log().Debug("chowning server root directory...")
 		if err := s.Filesystem().Chown("/"); err != nil {
 			return errors.WithMessage(err, "failed to chown root server directory during pre-boot process")
 		}
 	}
 
+	s.Log().Info("completed server preflight, starting boot process...")
 	return nil
 }

server/server.go

@@ -70,10 +70,10 @@ type Server struct {
 	wsBag       *WebsocketBag
 	wsBagLocker sync.Mutex
 
-	sinks map[SinkName]*sinkPool
+	sinks map[system.SinkName]*system.SinkPool
 
-	logSink     *sinkPool
-	installSink *sinkPool
+	logSink     *system.SinkPool
+	installSink *system.SinkPool
 }
 
 // New returns a new server instance with a context and all of the default
@@ -88,9 +88,9 @@ func New(client remote.Client) (*Server, error) {
 		transferring: system.NewAtomicBool(false),
 		restoring:    system.NewAtomicBool(false),
 		powerLock:    system.NewLocker(),
-		sinks: map[SinkName]*sinkPool{
-			LogSink:     newSinkPool(),
-			InstallSink: newSinkPool(),
+		sinks: map[system.SinkName]*system.SinkPool{
+			system.LogSink:     system.NewSinkPool(),
+			system.InstallSink: system.NewSinkPool(),
 		},
 	}
 	if err := defaults.Set(&s); err != nil {

server/update.go

@@ -1,6 +1,8 @@
 package server
 
 import (
+	"time"
+
 	"github.com/pterodactyl/wings/environment/docker"
 
 	"github.com/pterodactyl/wings/environment"
@@ -58,7 +60,7 @@ func (s *Server) SyncWithEnvironment() {
 			s.Log().Info("server suspended with running process state, terminating now")
 
 			go func(s *Server) {
-				if err := s.Environment.WaitForStop(60, true); err != nil {
+				if err := s.Environment.WaitForStop(s.Context(), time.Minute, true); err != nil {
 					s.Log().WithField("error", err).Warn("failed to terminate server environment after suspension")
 				}
 			}(s)

sftp/handler.go

@@ -288,14 +288,10 @@ func (h *Handler) can(permission string) bool {
 		return false
 	}
 
-	// SFTPServer owners and super admins have their permissions returned as '[*]' via the Panel
-	// API, so for the sake of speed do an initial check for that before iterating over the
-	// entire array of permissions.
-	if len(h.permissions) == 1 && h.permissions[0] == "*" {
-		return true
-	}
 	for _, p := range h.permissions {
-		if p == permission {
+		// If we match the permission specifically, or the user has been granted the "*"
+		// permission because they're an admin, let them through.
+		if p == permission || p == "*" {
 			return true
 		}
 	}

sftp/server.go

@@ -68,9 +68,14 @@ func (c *SFTPServer) Run() error {
 	}
 
 	conf := &ssh.ServerConfig{
-		NoClientAuth:     false,
-		MaxAuthTries:     6,
-		PasswordCallback: c.passwordCallback,
+		NoClientAuth: false,
+		MaxAuthTries: 6,
+		PasswordCallback: func(conn ssh.ConnMetadata, password []byte) (*ssh.Permissions, error) {
+			return c.makeCredentialsRequest(conn, remote.SftpAuthPassword, string(password))
+		},
+		PublicKeyCallback: func(conn ssh.ConnMetadata, key ssh.PublicKey) (*ssh.Permissions, error) {
+			return c.makeCredentialsRequest(conn, remote.SftpAuthPublicKey, string(ssh.MarshalAuthorizedKey(key)))
+		},
 	}
 	conf.AddHostKey(private)
 
@@ -177,17 +182,17 @@ func (c *SFTPServer) generateED25519PrivateKey() error {
 	return nil
 }
 
-// A function capable of validating user credentials with the Panel API.
-func (c *SFTPServer) passwordCallback(conn ssh.ConnMetadata, pass []byte) (*ssh.Permissions, error) {
+func (c *SFTPServer) makeCredentialsRequest(conn ssh.ConnMetadata, t remote.SftpAuthRequestType, p string) (*ssh.Permissions, error) {
 	request := remote.SftpAuthRequest{
+		Type:          t,
 		User:          conn.User(),
-		Pass:          string(pass),
+		Pass:          p,
 		IP:            conn.RemoteAddr().String(),
 		SessionID:     conn.SessionID(),
 		ClientVersion: conn.ClientVersion(),
 	}
 
-	logger := log.WithFields(log.Fields{"subsystem": "sftp", "username": conn.User(), "ip": conn.RemoteAddr().String()})
+	logger := log.WithFields(log.Fields{"subsystem": "sftp", "method": request.Type, "username": request.User, "ip": request.IP})
 	logger.Debug("validating credentials for SFTP connection")
 
 	if !validUsernameRegexp.MatchString(request.User) {
@@ -206,7 +211,7 @@ func (c *SFTPServer) passwordCallback(conn ssh.ConnMetadata, pass []byte) (*ssh.
 	}
 
 	logger.WithField("server", resp.Server).Debug("credentials validated and matched to server instance")
-	sshPerm := &ssh.Permissions{
+	permissions := ssh.Permissions{
 		Extensions: map[string]string{
 			"uuid":        resp.Server,
 			"user":        conn.User(),
@@ -214,7 +219,7 @@ func (c *SFTPServer) passwordCallback(conn ssh.ConnMetadata, pass []byte) (*ssh.
 		},
 	}
 
-	return sshPerm, nil
+	return &permissions, nil
 }
 
 // PrivateKeyPath returns the path the host private key for this server instance.

system/locker.go

@@ -42,7 +42,6 @@ func (l *Locker) Acquire() error {
 	return nil
 }
 
-
 // TryAcquire will attempt to acquire a power-lock until the context provided
 // is canceled.
 func (l *Locker) TryAcquire(ctx context.Context) error {
@@ -51,7 +50,9 @@ func (l *Locker) TryAcquire(ctx context.Context) error {
 		return nil
 	case <-ctx.Done():
 		if err := ctx.Err(); err != nil {
-			return err
+			if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
+				return ErrLockerLocked
+			}
 		}
 		return nil
 	}

system/locker_test.go

@@ -81,7 +81,7 @@ func TestPower(t *testing.T) {
 				err := l.TryAcquire(ctx)
 
 				g.Assert(err).IsNotNil()
-				g.Assert(errors.Is(err, context.DeadlineExceeded)).IsTrue()
+				g.Assert(errors.Is(err, ErrLockerLocked)).IsTrue()
 				g.Assert(cap(l.ch)).Equal(1)
 				g.Assert(len(l.ch)).Equal(1)
 				g.Assert(l.IsLocked()).IsTrue()
@@ -95,7 +95,7 @@ func TestPower(t *testing.T) {
 
 				l.Acquire()
 				go func() {
-					time.AfterFunc(time.Millisecond * 50, func() {
+					time.AfterFunc(time.Millisecond*50, func() {
 						l.Release()
 					})
 				}()

system/rate.go

@@ -44,7 +44,7 @@ func (r *Rate) Try() bool {
 // Reset resets the internal state of the rate limiter back to zero.
 func (r *Rate) Reset() {
 	r.mu.Lock()
-    r.count = 0
-    r.last = time.Now()
+	r.count = 0
+	r.last = time.Now()
 	r.mu.Unlock()
 }

system/rate_test.go

@@ -47,7 +47,7 @@ func TestRate(t *testing.T) {
 		g.It("resets back to zero when called", func() {
 			r := NewRate(10, time.Second)
 			for i := 0; i < 100; i++ {
-				if i % 10 == 0 {
+				if i%10 == 0 {
 					r.Reset()
 				}
 				g.Assert(r.Try()).IsTrue()

system/sink_pool.go

@@ -1,4 +1,4 @@
-package server
+package system
 
 import (
 	"sync"
@@ -16,20 +16,20 @@ const (
 	InstallSink SinkName = "install"
 )
 
-// sinkPool represents a pool with sinks.
-type sinkPool struct {
+// SinkPool represents a pool with sinks.
+type SinkPool struct {
 	mu    sync.RWMutex
 	sinks []chan []byte
 }
 
-// newSinkPool returns a new empty sinkPool. A sink pool generally lives with a
+// NewSinkPool returns a new empty SinkPool. A sink pool generally lives with a
 // server instance for it's full lifetime.
-func newSinkPool() *sinkPool {
-	return &sinkPool{}
+func NewSinkPool() *SinkPool {
+	return &SinkPool{}
 }
 
 // On adds a channel to the sink pool instance.
-func (p *sinkPool) On(c chan []byte) {
+func (p *SinkPool) On(c chan []byte) {
 	p.mu.Lock()
 	p.sinks = append(p.sinks, c)
 	p.mu.Unlock()
@@ -37,7 +37,7 @@ func (p *sinkPool) On(c chan []byte) {
 
 // Off removes a given channel from the sink pool. If no matching sink is found
 // this function is a no-op. If a matching channel is found, it will be removed.
-func (p *sinkPool) Off(c chan []byte) {
+func (p *SinkPool) Off(c chan []byte) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
@@ -66,7 +66,7 @@ func (p *sinkPool) Off(c chan []byte) {
 
 // Destroy destroys the pool by removing and closing all sinks and destroying
 // all of the channels that are present.
-func (p *sinkPool) Destroy() {
+func (p *SinkPool) Destroy() {
 	p.mu.Lock()
 	defer p.mu.Unlock()
 
@@ -95,7 +95,7 @@ func (p *sinkPool) Destroy() {
 // likely the best option anyways. This uses waitgroups to allow every channel
 // to attempt its send concurrently thus making the total blocking time of this
 // function "O(1)" instead of "O(n)".
-func (p *sinkPool) Push(data []byte) {
+func (p *SinkPool) Push(data []byte) {
 	p.mu.RLock()
 	defer p.mu.RUnlock()
 	var wg sync.WaitGroup
@@ -119,24 +119,3 @@ func (p *sinkPool) Push(data []byte) {
 	}
 	wg.Wait()
 }
-
-// Sink returns the instantiated and named sink for a server. If the sink has
-// not been configured yet this function will cause a panic condition.
-func (s *Server) Sink(name SinkName) *sinkPool {
-	sink, ok := s.sinks[name]
-	if !ok {
-		s.Log().Fatalf("attempt to access nil sink: %s", name)
-	}
-	return sink
-}
-
-// DestroyAllSinks iterates over all of the sinks configured for the server and
-// destroys their instances. Note that this will cause a panic if you attempt
-// to call Server.Sink() again after. This function is only used when a server
-// is being deleted from the system.
-func (s *Server) DestroyAllSinks() {
-	s.Log().Info("destroying all registered sinks for server instance")
-	for _, sink := range s.sinks {
-		sink.Destroy()
-	}
-}

system/sink_pool_test.go

@@ -1,4 +1,4 @@
-package server
+package system
 
 import (
 	"fmt"
@@ -23,7 +23,7 @@ func TestSink(t *testing.T) {
 
 	g.Describe("SinkPool#On", func() {
 		g.It("pushes additional channels to a sink", func() {
-			pool := &sinkPool{}
+			pool := &SinkPool{}
 
 			g.Assert(pool.sinks).IsZero()
 
@@ -36,9 +36,9 @@ func TestSink(t *testing.T) {
 	})
 
 	g.Describe("SinkPool#Off", func() {
-		var pool *sinkPool
+		var pool *SinkPool
 		g.BeforeEach(func() {
-			pool = &sinkPool{}
+			pool = &SinkPool{}
 		})
 
 		g.It("works when no sinks are registered", func() {
@@ -97,9 +97,9 @@ func TestSink(t *testing.T) {
 	})
 
 	g.Describe("SinkPool#Push", func() {
-		var pool *sinkPool
+		var pool *SinkPool
 		g.BeforeEach(func() {
-			pool = &sinkPool{}
+			pool = &SinkPool{}
 		})
 
 		g.It("works when no sinks are registered", func() {
@@ -190,9 +190,9 @@ func TestSink(t *testing.T) {
 	})
 
 	g.Describe("SinkPool#Destroy", func() {
-		var pool *sinkPool
+		var pool *SinkPool
 		g.BeforeEach(func() {
-			pool = &sinkPool{}
+			pool = &SinkPool{}
 		})
 
 		g.It("works if no sinks are registered", func() {

system/strings.go

@@ -0,0 +1,20 @@
+package system
+
+import (
+	"math/rand"
+	"strings"
+)
+
+const characters = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890"
+
+// RandomString generates a random string of alpha-numeric characters using a
+// pseudo-random number generator. The output of this function IS NOT cryptographically
+// secure, it is used solely for generating random strings outside a security context.
+func RandomString(n int) string {
+	var b strings.Builder
+	b.Grow(n)
+	for i := 0; i < n; i++ {
+		b.WriteByte(characters[rand.Intn(len(characters))])
+	}
+	return b.String()
+}

system/utils.go

@@ -3,12 +3,10 @@ package system
 import (
 	"bufio"
 	"bytes"
-	"context"
 	"fmt"
 	"io"
 	"strconv"
 	"sync"
-	"time"
 
 	"emperror.dev/errors"
 	"github.com/goccy/go-json"
@@ -90,16 +88,16 @@ func ScanReader(r io.Reader, callback func(line []byte)) error {
 			} else {
 				buf.Write(line)
 			}
+			// If we encountered an error with something in ReadLine that was not an
+			// EOF just abort the entire process here.
+			if err != nil && err != io.EOF {
+				return err
+			}
 			// Finish this loop and begin outputting the line if there is no prefix
 			// (the line fit into the default buffer), or if we hit the end of the line.
 			if !isPrefix || err == io.EOF {
 				break
 			}
-			// If we encountered an error with something in ReadLine that was not an
-			// EOF just abort the entire process here.
-			if err != nil {
-				return err
-			}
 		}
 
 		// Send the full buffer length over to the event handler to be emitted in
@@ -122,22 +120,6 @@ func ScanReader(r io.Reader, callback func(line []byte)) error {
 	return nil
 }
 
-// Runs a given work function every "d" duration until the provided context is canceled.
-func Every(ctx context.Context, d time.Duration, work func(t time.Time)) {
-	ticker := time.NewTicker(d)
-	go func() {
-		for {
-			select {
-			case <-ctx.Done():
-				ticker.Stop()
-				return
-			case t := <-ticker.C:
-				work(t)
-			}
-		}
-	}()
-}
-
 func FormatBytes(b int64) string {
 	if b < 1024 {
 		return fmt.Sprintf("%d B", b)

wings.go

@@ -2,8 +2,16 @@ package main
 
 import (
 	"github.com/pterodactyl/wings/cmd"
+	"math/rand"
+	"time"
 )
 
 func main() {
+	// Since we make use of the math/rand package in the code, especially for generating
+	// non-cryptographically secure random strings we need to seed the RNG. Just make use
+	// of the current time for this.
+	rand.Seed(time.Now().UnixNano())
+
+	// Execute the main binary code.
 	cmd.Execute()
 }