runner/main.go

376 lines
9.6 KiB
Go
Raw Normal View History

2024-02-03 04:17:32 +01:00
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())
}
}