runner/main.go

package main

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"

	"gitea.larvit.se/pwrpln/go_log"
	"github.com/google/uuid"
	"github.com/joho/godotenv"
)

type Payload struct {
	Commands []string `json:"commands"`
}

type User struct {
	HomeDir  string
	Username string
}

func cleanEnv(log go_log.Log, workDir string, uid int, gid int) (string, error) {
	response := ""

	log.Debug("Creating working directory if it does not exist", "workDir", workDir)
	response += "Creating working directory if it does not exist\n"
	err := os.MkdirAll(workDir, os.ModePerm)
	if err != nil {
		log.Error("Could not create working directory", err, err.Error())
		return response, err
	}

	log.Debug("Resetting owner on working directory")
	response += "Resetting owner on working directory"
	err = os.Chown(workDir, uid, gid)
	if err != nil {
		log.Error("Could not set owner on working directory", err, err.Error())
		return response, err
	}

	log.Debug("Cleaning up previous files and/or folders in working directory", "workDir", workDir)
	response += "Cleaning up previous files and/or folders in working directory\n"
	err = rmDirectoryContent(log, workDir)
	if err != nil {
		log.Error("Could not clear workDir")
		return response, err
	}

	// Remove all docker containeres
	_, _, err = runCmd(log, workDir, "docker ps -aq | xargs -r docker rm -f", uid, gid)
	response += "Removing all pre-existing docker containers\n"
	if err != nil {
		log.Error("Could not remove all containers", "err", err.Error())
		return response, err
	}

	// Remove all docker networks
	_, _, err = runCmd(log, workDir, "docker network prune --force", uid, gid)
	response += "Removing all pre-existing docker networks\n"
	if err != nil {
		log.Error("Could not remove all docker networks", "err", err.Error())
		return response, err
	}

	// Remove all docker volumes
	_, _, err = runCmd(log, workDir, "docker volume prune -a -f", uid, gid)
	response += "Removing all pre-existing docker volumes\n"
	if err != nil {
		log.Error("Could not remove all docker volumes", "err", err.Error())
		return response, err
	}

	// Log out from all docker registries
	_, _, err = runCmd(log, workDir, "for url in $(cat ~/.docker/config.json | jq -r '.auths | keys | .[]'); do docker logout $url; done", uid, gid)
	response += "Logging out from all configured docker registries\n"
	if err != nil {
		log.Error("Could not log out from all configured docker registries", "err", err.Error())
		return response, err
	}

	return response, nil
}

func getUserByUid(log go_log.Log, uid int) (User, error) {
	// Get uids home directory
	passwdFile, err := os.Open("/etc/passwd")
	if err != nil {
		log.Error("Could not read /etc/passwd", "err", err.Error())
		return User{}, err
	}
	defer passwdFile.Close()

	// Read file line by line
	scanner := bufio.NewScanner(passwdFile)
	for scanner.Scan() {
		parts := strings.Split(scanner.Text(), ":")
		if parts[2] == strconv.Itoa(uid) {
			return User{
				HomeDir:  parts[5],
				Username: parts[0],
			}, nil
		}
	}

	err = scanner.Err()
	if err != nil {
		log.Error("Could not scan /etc/passwd", "err", err.Error())
		return User{}, err
	}

	log.Error("Could not find users home directory in /etc/passwd", "uid", uid)
	err = errors.New("could not find users home directory in /etc/passwd")
	return User{}, err
}

func rmDirectoryContent(log go_log.Log, dir string) error {
	d, err := os.Open(dir)
	if err != nil {
		log.Error("Could not open directory", "dir", dir, "err", err.Error())
		return err
	}
	defer d.Close()
	names, err := d.Readdirnames(-1)
	if err != nil {
		log.Error("Could not list names in dir", "dir", dir, "err", err.Error())
		return err
	}
	for _, name := range names {
		err = os.RemoveAll(filepath.Join(dir, name))
		if err != nil {
			log.Error("Could not remove path", "path", filepath.Join(dir, name), "err", err.Error())
			return err
		}
	}
	return nil
}

func runCmd(log go_log.Log, workDir string, cmdStr string, uid int, gid int) (int, string, error) {
	response := ""

	user, err := getUserByUid(log, uid)
	if err != nil {
		return 1, response, err
	}

	// Create the command
	cmd := exec.Command("/bin/sh", "-c", cmdStr)
	cmd.Dir = workDir
	cmd.SysProcAttr = &syscall.SysProcAttr{}
	cmd.SysProcAttr.Credential = &syscall.Credential{Uid: uint32(uid), Gid: uint32(gid)}

	cmd.Env = append(cmd.Env, "USER="+user.Username)
	cmd.Env = append(cmd.Env, "HOME="+user.HomeDir)

	// Create a pipes to capture the command's output
	stdoutPipe, err := cmd.StdoutPipe()
	if err != nil {
		log.Error("Could not create stdout pipe from command", "err", err.Error())
		return 1, response, err
	}
	stderrPipe, err := cmd.StderrPipe()
	if err != nil {
		log.Error("Could not create stderr pipe from command", "err", err.Error())
		return 1, response, err
	}

	// Start the command
	err = cmd.Start()
	if err != nil {
		log.Error("Could not start the command", "err", err.Error())
		return 1, response, err
	}

	defer stdoutPipe.Close()
	defer stderrPipe.Close()

	var wg sync.WaitGroup

	log.Debug("Running command", "cmdStr", cmdStr)

	wg.Add(1)
	go func() {
		buf := make([]byte, 1024)
		for {
			n, err := stdoutPipe.Read(buf)
			if err == io.EOF {
				break
			}
			if err != nil {
				log.Error("Error reading stdout from command", "err", err.Error())
				break
			}
			log.Debug("runCmd() stdOut", "buf", string(buf[:n]))
			response += "stdOut: " + string(buf[:n])
		}
		wg.Done()
	}()

	wg.Add(1)
	go func() {
		buf := make([]byte, 1024)
		for {
			n, err := stderrPipe.Read(buf)
			if err == io.EOF {
				break
			}
			if err != nil {
				log.Error("Error reading stderr from command", "err", err.Error())
				break
			}
			log.Debug("runCmd() stdErr", "buf", string(buf[:n]))
			response += "stdErr: " + string(buf[:n])
		}
		wg.Done()
	}()

	wg.Wait()

	err = cmd.Wait()
	if err != nil {
		exitErr, ok := err.(*exec.ExitError)
		if !ok {
			log.Error("Fatal error on cmd.Wait()", "err", err.Error())
			return 1, response, err
		}

		return exitErr.ExitCode(), response, nil
	}

	return 0, response, nil
}

func setRunnerFree(runnerActive *bool) {
	*runnerActive = false
}

func main() {
	runnerActive := false

	// Load ENV from .env file
	err := godotenv.Load()
	if err != nil {
		tmpLog := go_log.GetLog()
		tmpLog.Info("Failed to load .env file!")
	}

	acceptedApiKeys := strings.Split(os.Getenv("API_KEYS"), ",")
	httpPort := os.Getenv("PORT")
	logLevel := os.Getenv("LOG_LEVEL")
	logLocation := os.Getenv("LOG_LOCATION")
	workDir := os.Getenv("WORK_DIR")

	// Setting up logger
	mainLog := go_log.GetLog()
	mainLog.MinLogLvl = go_log.LogLvlFromStr(logLevel)
	if mainLog.MinLogLvl == 0 {
		mainLog.MinLogLvl = 3 // Fall back to Info log level, so we don't miss important errors
	}
	loc, err := time.LoadLocation(logLocation)
	if err != nil {
		panic(err)
	}
	mainLog.TimeLocation = loc

	uid, err := strconv.Atoi(os.Getenv("RUN_CMD_AS_UID"))
	if err != nil {
		mainLog.Error("Could not read RUN_CMD_AS_UID as integer", "RUN_CMD_AS_UID", os.Getenv("RUN_CMD_AS_UID"), "err", err.Error())
		os.Exit(1)
	}

	gid, err := strconv.Atoi(os.Getenv("RUN_CMD_AS_GID"))
	if err != nil {
		mainLog.Error("Could not read RUN_CMD_AS_GID as integer", "RUN_CMD_AS_GID", os.Getenv("RUN_CMD_AS_GID"), "err", err.Error())
		os.Exit(1)
	}

	mainLog.Info(
		"Configuration is determined",
		"API_KEYS", len(acceptedApiKeys),
		"LOG_LEVEL", logLevel,
		"LOG_LOCATION", logLocation,
		"WORK_DIR", workDir,
		"PORT", httpPort,
		"RUN_CMD_AS_GID", gid,
		"RUN_CMD_AS_UID", uid,
	)

	http.HandleFunc("/", func(res http.ResponseWriter, req *http.Request) {
		reqId := uuid.NewString()
		log := mainLog // Make copy of main log to use in each request
		log.Context = append(log.Context, "reqId", reqId)
		defer log.Verbose("Run complete")
		response := ""
		providedApiKey := req.Header.Get("Authorization")

		if !slices.Contains(acceptedApiKeys, providedApiKey) {
			log.Debug("Invalid API key provided")
			res.WriteHeader(401)
			res.Write([]byte("Unauthorized"))
			return
		}

		log.Verbose("Making a run")

		for runnerActive {
			log.Debug("Another runner is running, waiting")
			time.Sleep(1 * time.Second)
		}
		runnerActive = true
		defer setRunnerFree(&runnerActive)

		// Clean up environment
		cleanEnvRes, err := cleanEnv(log, workDir, uid, gid)
		if err != nil {
			res.WriteHeader(500)
			res.Write([]byte("Internal Server Error"))
			return
		}
		response += cleanEnvRes

		decoder := json.NewDecoder(req.Body)
		var payload Payload
		err = decoder.Decode(&payload)
		if err != nil {
			log.Verbose("Invalid JSON payload", "err", err.Error())
			res.WriteHeader(400)
			res.Write([]byte("Invalid JSON payload, err: \"" + err.Error() + "\""))
			return
		}

		res.Header().Set("Content-Type", "text/event-stream")
		res.Header().Set("Cache-Control", "no-cache")
		res.Header().Set("Connection", "keep-alive")

		for idx, command := range payload.Commands {
			response += "Running command #" + fmt.Sprint(idx) + ": " + command + "\n"
			exitCode, cmdRes, err := runCmd(log, workDir, command, uid, gid)
			response += cmdRes
			if err != nil {
				response += "Command failed, err: " + err.Error() + "\n"
				res.WriteHeader(400)
				res.Write([]byte(response))
				return
			}
			if exitCode != 0 {
				response += "Command failed, non-zero exit code: " + fmt.Sprint(exitCode) + "\n"
				res.WriteHeader(400)
				res.Write([]byte(response))
				return
			}
		}

		// If we get here, nothing have broken down, so 200 OK and send the full response
		res.WriteHeader(200)
		res.Write([]byte(response))
	})

	mainLog.Info("Starting web server", "PORT", httpPort)
	err = http.ListenAndServe(":"+httpPort, nil)
	if err != nil {
		mainLog.Error("Web server crashed", "err", err.Error())
	}
}