executor.go 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878
  1. package executor
  2. import (
  3. acl "github.com/OliveTin/OliveTin/internal/acl"
  4. "github.com/OliveTin/OliveTin/internal/auth"
  5. authpublic "github.com/OliveTin/OliveTin/internal/auth/authpublic"
  6. config "github.com/OliveTin/OliveTin/internal/config"
  7. "github.com/OliveTin/OliveTin/internal/entities"
  8. "github.com/google/uuid"
  9. log "github.com/sirupsen/logrus"
  10. "github.com/prometheus/client_golang/prometheus"
  11. "github.com/prometheus/client_golang/prometheus/promauto"
  12. "gopkg.in/yaml.v3"
  13. "bytes"
  14. "context"
  15. "fmt"
  16. "os"
  17. "os/exec"
  18. "path"
  19. "strings"
  20. "sync"
  21. "time"
  22. )
  23. const (
  24. DefaultExitCodeNotExecuted = -1337
  25. MaxTriggerDepth = 10
  26. )
  27. var (
  28. metricActionsRequested = promauto.NewCounter(prometheus.CounterOpts{
  29. Name: "olivetin_actions_requested_count",
  30. Help: "The actions requested count",
  31. })
  32. )
  33. type ActionBinding struct {
  34. ID string
  35. Action *config.Action
  36. Entity *entities.Entity
  37. ConfigOrder int
  38. IsOnDashboard bool
  39. }
  40. // Executor represents a helper class for executing commands. It's main method
  41. // is ExecRequest
  42. type Executor struct {
  43. logs map[string]*InternalLogEntry
  44. logsTrackingIdsByDate []string
  45. LogsByActionId map[string][]*InternalLogEntry
  46. logmutex sync.RWMutex
  47. MapActionIdToBinding map[string]*ActionBinding
  48. MapActionIdToBindingLock sync.RWMutex
  49. Cfg *config.Config
  50. listeners []listener
  51. chainOfCommand []executorStepFunc
  52. }
  53. // ExecutionRequest is a request to execute an action. It's passed to an
  54. // Executor. They're created from the api.
  55. type ExecutionRequest struct {
  56. Binding *ActionBinding
  57. Arguments map[string]string
  58. TrackingID string
  59. Tags []string
  60. Cfg *config.Config
  61. AuthenticatedUser *authpublic.AuthenticatedUser
  62. TriggerDepth int
  63. logEntry *InternalLogEntry
  64. finalParsedCommand string
  65. execArgs []string
  66. useDirectExec bool
  67. executor *Executor
  68. }
  69. // InternalLogEntry objects are created by an Executor, and represent the final
  70. // state of execution (even if the command is not executed). It's designed to be
  71. // easily serializable.
  72. type InternalLogEntry struct {
  73. Binding *ActionBinding
  74. BindingID string
  75. DatetimeStarted time.Time
  76. DatetimeFinished time.Time
  77. Output string
  78. TimedOut bool
  79. Blocked bool
  80. ExitCode int32
  81. Tags []string
  82. ExecutionStarted bool
  83. ExecutionFinished bool
  84. ExecutionTrackingID string
  85. Process *os.Process
  86. Username string
  87. Index int64
  88. EntityPrefix string
  89. ActionConfigTitle string // This is the title of the action as defined in the config, not the final parsed title.
  90. /*
  91. The following 3 properties are obviously on Action normally, but it's useful
  92. that logs are lightweight (so we don't need to have an action associated to
  93. logs, etc. Therefore, we duplicate those values here.
  94. */
  95. ActionTitle string
  96. ActionIcon string
  97. ActionId string
  98. }
  99. type executorStepFunc func(*ExecutionRequest) bool
  100. // DefaultExecutor returns an Executor, with a sensible "chain of command" for
  101. // executing actions.
  102. func DefaultExecutor(cfg *config.Config) *Executor {
  103. e := Executor{}
  104. e.Cfg = cfg
  105. e.logs = make(map[string]*InternalLogEntry)
  106. e.logsTrackingIdsByDate = make([]string, 0)
  107. e.LogsByActionId = make(map[string][]*InternalLogEntry)
  108. e.MapActionIdToBinding = make(map[string]*ActionBinding)
  109. e.chainOfCommand = []executorStepFunc{
  110. stepRequestAction,
  111. stepConcurrencyCheck,
  112. stepRateCheck,
  113. stepACLCheck,
  114. stepParseArgs,
  115. stepLogStart,
  116. stepExec,
  117. stepExecAfter,
  118. stepLogFinish,
  119. stepSaveLog,
  120. stepTrigger,
  121. }
  122. return &e
  123. }
  124. type listener interface {
  125. OnExecutionStarted(logEntry *InternalLogEntry)
  126. OnExecutionFinished(logEntry *InternalLogEntry)
  127. OnOutputChunk(o []byte, executionTrackingId string)
  128. OnActionMapRebuilt()
  129. }
  130. func (e *Executor) AddListener(m listener) {
  131. e.listeners = append(e.listeners, m)
  132. }
  133. // getPagingStartIndex calculates the starting index for log pagination.
  134. // Parameters:
  135. //
  136. // startOffset: The offset from the most recent log (0 means start from the most recent)
  137. // totalLogCount: Total number of logs available
  138. // count: Number of logs to retrieve
  139. //
  140. // Returns: The calculated starting index for pagination
  141. func getPagingStartIndex(startOffset int64, totalLogCount int64) int64 {
  142. var startIndex int64
  143. if startOffset <= 0 {
  144. startIndex = totalLogCount
  145. } else {
  146. startIndex = (totalLogCount - startOffset)
  147. if startIndex < 0 {
  148. startIndex = 1
  149. }
  150. }
  151. return startIndex - 1
  152. }
  153. type PagingResult struct {
  154. CountRemaining int64
  155. PageSize int64
  156. TotalCount int64
  157. StartOffset int64
  158. }
  159. func (e *Executor) GetLogTrackingIds(startOffset int64, pageCount int64) ([]*InternalLogEntry, *PagingResult) {
  160. pagingResult := &PagingResult{
  161. CountRemaining: 0,
  162. PageSize: pageCount,
  163. TotalCount: 0,
  164. StartOffset: startOffset,
  165. }
  166. e.logmutex.RLock()
  167. totalLogCount := int64(len(e.logsTrackingIdsByDate))
  168. pagingResult.TotalCount = totalLogCount
  169. startIndex := getPagingStartIndex(startOffset, totalLogCount)
  170. pageCount = min(totalLogCount, pageCount)
  171. endIndex := max(0, (startIndex-pageCount)+1)
  172. log.WithFields(log.Fields{
  173. "startOffset": startOffset,
  174. "pageCount": pageCount,
  175. "total": totalLogCount,
  176. "startIndex": startIndex,
  177. "endIndex": endIndex,
  178. }).Tracef("GetLogTrackingIds")
  179. trackingIds := make([]*InternalLogEntry, 0, pageCount)
  180. if totalLogCount > 0 {
  181. for i := endIndex; i <= startIndex; i++ {
  182. trackingIds = append(trackingIds, e.logs[e.logsTrackingIdsByDate[i]])
  183. }
  184. }
  185. e.logmutex.RUnlock()
  186. pagingResult.CountRemaining = endIndex
  187. return trackingIds, pagingResult
  188. }
  189. // isValidLogEntryForACL checks if a log entry has all required fields for ACL checking.
  190. func isValidLogEntryForACL(entry *InternalLogEntry) bool {
  191. return entry != nil && entry.Binding != nil && entry.Binding.Action != nil
  192. }
  193. // isLogEntryAllowedByACL checks if a log entry is allowed to be viewed by the user.
  194. func isLogEntryAllowedByACL(cfg *config.Config, user *authpublic.AuthenticatedUser, entry *InternalLogEntry) bool {
  195. return acl.IsAllowedLogs(cfg, user, entry.Binding.Action)
  196. }
  197. func (e *Executor) filterLogsByACL(cfg *config.Config, user *authpublic.AuthenticatedUser) []*InternalLogEntry {
  198. e.logmutex.RLock()
  199. defer e.logmutex.RUnlock()
  200. filtered := make([]*InternalLogEntry, 0, len(e.logsTrackingIdsByDate))
  201. for _, trackingId := range e.logsTrackingIdsByDate {
  202. entry := e.logs[trackingId]
  203. if !isValidLogEntryForACL(entry) {
  204. continue
  205. }
  206. if isLogEntryAllowedByACL(cfg, user, entry) {
  207. filtered = append(filtered, entry)
  208. }
  209. }
  210. return filtered
  211. }
  212. // paginateFilteredLogs applies pagination to a filtered list of logs and returns
  213. // the paginated results along with pagination metadata.
  214. func paginateFilteredLogs(filtered []*InternalLogEntry, startOffset int64, pageCount int64) ([]*InternalLogEntry, *PagingResult) {
  215. total := int64(len(filtered))
  216. paging := &PagingResult{PageSize: pageCount, TotalCount: total, StartOffset: startOffset}
  217. if total == 0 {
  218. paging.CountRemaining = 0
  219. return []*InternalLogEntry{}, paging
  220. }
  221. startIndex := getPagingStartIndex(startOffset, total)
  222. pageCount = min(total, pageCount)
  223. endIndex := max(0, (startIndex-pageCount)+1)
  224. out := make([]*InternalLogEntry, 0, pageCount)
  225. for i := endIndex; i <= startIndex && i < int64(len(filtered)); i++ {
  226. out = append(out, filtered[i])
  227. }
  228. paging.CountRemaining = endIndex
  229. return out, paging
  230. }
  231. // GetLogTrackingIdsACL returns logs filtered by ACL visibility for the user and
  232. // paginated correctly based on the filtered set.
  233. func (e *Executor) GetLogTrackingIdsACL(cfg *config.Config, user *authpublic.AuthenticatedUser, startOffset int64, pageCount int64) ([]*InternalLogEntry, *PagingResult) {
  234. filtered := e.filterLogsByACL(cfg, user)
  235. return paginateFilteredLogs(filtered, startOffset, pageCount)
  236. }
  237. func (e *Executor) GetLog(trackingID string) (*InternalLogEntry, bool) {
  238. e.logmutex.RLock()
  239. entry, found := e.logs[trackingID]
  240. e.logmutex.RUnlock()
  241. return entry, found
  242. }
  243. func (e *Executor) GetLogsByActionId(actionId string) []*InternalLogEntry {
  244. e.logmutex.RLock()
  245. logs, found := e.LogsByActionId[actionId]
  246. e.logmutex.RUnlock()
  247. if !found {
  248. return make([]*InternalLogEntry, 0)
  249. }
  250. return logs
  251. }
  252. func (e *Executor) SetLog(trackingID string, entry *InternalLogEntry) {
  253. e.logmutex.Lock()
  254. entry.Index = int64(len(e.logsTrackingIdsByDate))
  255. e.logs[trackingID] = entry
  256. e.logsTrackingIdsByDate = append(e.logsTrackingIdsByDate, trackingID)
  257. e.logmutex.Unlock()
  258. }
  259. // ExecRequest processes an ExecutionRequest
  260. func (e *Executor) ExecRequest(req *ExecutionRequest) (*sync.WaitGroup, string) {
  261. if req.AuthenticatedUser == nil {
  262. req.AuthenticatedUser = auth.UserGuest(req.Cfg)
  263. }
  264. req.executor = e
  265. req.logEntry = &InternalLogEntry{
  266. Binding: req.Binding,
  267. DatetimeStarted: time.Now(),
  268. ExecutionTrackingID: req.TrackingID,
  269. Output: "",
  270. ExitCode: DefaultExitCodeNotExecuted,
  271. ExecutionStarted: false,
  272. ExecutionFinished: false,
  273. ActionId: "",
  274. ActionTitle: "notfound",
  275. ActionIcon: "&#x1f4a9;",
  276. Username: req.AuthenticatedUser.Username,
  277. }
  278. _, isDuplicate := e.GetLog(req.TrackingID)
  279. if isDuplicate || req.TrackingID == "" {
  280. req.TrackingID = uuid.NewString()
  281. }
  282. // Update the log entry with the final tracking ID
  283. req.logEntry.ExecutionTrackingID = req.TrackingID
  284. log.Tracef("executor.ExecRequest(): %v", req)
  285. e.SetLog(req.TrackingID, req.logEntry)
  286. wg := new(sync.WaitGroup)
  287. wg.Add(1)
  288. go func() {
  289. e.execChain(req)
  290. defer wg.Done()
  291. }()
  292. return wg, req.TrackingID
  293. }
  294. func (e *Executor) execChain(req *ExecutionRequest) {
  295. for _, step := range e.chainOfCommand {
  296. if !step(req) {
  297. break
  298. }
  299. }
  300. req.logEntry.ExecutionFinished = true
  301. // This isn't a step, because we want to notify all listeners, irrespective
  302. // of how many steps were actually executed.
  303. notifyListenersFinished(req)
  304. }
  305. func getConcurrentCount(req *ExecutionRequest) int {
  306. concurrentCount := 0
  307. req.executor.logmutex.RLock()
  308. for _, log := range req.executor.GetLogsByActionId(req.Binding.Action.ID) {
  309. if !log.ExecutionFinished {
  310. concurrentCount += 1
  311. }
  312. }
  313. req.executor.logmutex.RUnlock()
  314. return concurrentCount
  315. }
  316. func stepConcurrencyCheck(req *ExecutionRequest) bool {
  317. concurrentCount := getConcurrentCount(req)
  318. // Note that the current execution is counted int the logs, so when checking we +1
  319. if concurrentCount >= (req.Binding.Action.MaxConcurrent + 1) {
  320. log.WithFields(log.Fields{
  321. "actionTitle": req.logEntry.ActionTitle,
  322. "concurrentCount": concurrentCount,
  323. "maxConcurrent": req.Binding.Action.MaxConcurrent,
  324. }).Warnf("Blocked from executing due to concurrency limit")
  325. req.logEntry.Output = "Blocked from executing due to concurrency limit"
  326. req.logEntry.Blocked = true
  327. return false
  328. }
  329. return true
  330. }
  331. func parseDuration(rate config.RateSpec) time.Duration {
  332. duration, err := time.ParseDuration(rate.Duration)
  333. if err != nil {
  334. log.Warnf("Could not parse duration: %v", rate.Duration)
  335. return -1 * time.Minute
  336. }
  337. return duration
  338. }
  339. //gocyclo:ignore
  340. func getExecutionsCount(rate config.RateSpec, req *ExecutionRequest) int {
  341. executions := -1 // Because we will find ourself when checking execution logs
  342. duration := parseDuration(rate)
  343. then := time.Now().Add(-duration)
  344. for _, logEntry := range req.executor.GetLogsByActionId(req.Binding.Action.ID) {
  345. // FIXME
  346. /*
  347. if logEntry.EntityPrefix != req.EntityPrefix {
  348. continue
  349. }
  350. */
  351. if logEntry.DatetimeStarted.After(then) && !logEntry.Blocked {
  352. executions += 1
  353. }
  354. }
  355. return executions
  356. }
  357. func stepRateCheck(req *ExecutionRequest) bool {
  358. for _, rate := range req.Binding.Action.MaxRate {
  359. executions := getExecutionsCount(rate, req)
  360. if executions >= rate.Limit {
  361. log.WithFields(log.Fields{
  362. "actionTitle": req.logEntry.ActionTitle,
  363. "executions": executions,
  364. "limit": rate.Limit,
  365. "duration": rate.Duration,
  366. }).Infof("Blocked from executing due to rate limit")
  367. req.logEntry.Output = "Blocked from executing due to rate limit"
  368. req.logEntry.Blocked = true
  369. return false
  370. }
  371. }
  372. return true
  373. }
  374. func stepACLCheck(req *ExecutionRequest) bool {
  375. canExec := acl.IsAllowedExec(req.Cfg, req.AuthenticatedUser, req.Binding.Action)
  376. if !canExec {
  377. req.logEntry.Output = "ACL check failed. Blocked from executing."
  378. req.logEntry.Blocked = true
  379. log.WithFields(log.Fields{
  380. "actionTitle": req.logEntry.ActionTitle,
  381. }).Warnf("ACL check failed. Blocked from executing.")
  382. }
  383. return canExec
  384. }
  385. func stepParseArgs(req *ExecutionRequest) bool {
  386. ensureArgumentMap(req)
  387. injectSystemArgs(req)
  388. if !hasBindingAndAction(req) {
  389. return fail(req, fmt.Errorf("cannot parse arguments: Binding or Action is nil"))
  390. }
  391. mangleInvalidArgumentValues(req)
  392. if hasExec(req) {
  393. return handleExecBranch(req)
  394. } else {
  395. return handleShellBranch(req)
  396. }
  397. }
  398. func handleExecBranch(req *ExecutionRequest) bool {
  399. args, err := parseActionExec(req.Arguments, req.Binding.Action, req.Binding.Entity)
  400. if err != nil {
  401. return fail(req, err)
  402. }
  403. req.useDirectExec = true
  404. req.execArgs = args
  405. return true
  406. }
  407. func handleShellBranch(req *ExecutionRequest) bool {
  408. if err := checkShellArgumentSafety(req.Binding.Action); err != nil {
  409. return fail(req, err)
  410. }
  411. cmd, err := parseActionArguments(req.Arguments, req.Binding.Action, req.Binding.Entity)
  412. if err != nil {
  413. return fail(req, err)
  414. }
  415. req.useDirectExec = false
  416. req.finalParsedCommand = cmd
  417. return true
  418. }
  419. func ensureArgumentMap(req *ExecutionRequest) {
  420. if req.Arguments == nil {
  421. req.Arguments = make(map[string]string)
  422. }
  423. }
  424. func injectSystemArgs(req *ExecutionRequest) {
  425. req.Arguments["ot_executionTrackingId"] = req.TrackingID
  426. req.Arguments["ot_username"] = req.AuthenticatedUser.Username
  427. }
  428. func hasBindingAndAction(req *ExecutionRequest) bool {
  429. return !(req.Binding == nil || req.Binding.Action == nil)
  430. }
  431. func hasExec(req *ExecutionRequest) bool {
  432. return len(req.Binding.Action.Exec) > 0
  433. }
  434. func fail(req *ExecutionRequest, err error) bool {
  435. req.logEntry.Output = err.Error()
  436. log.Warn(err.Error())
  437. return false
  438. }
  439. func stepRequestAction(req *ExecutionRequest) bool {
  440. metricActionsRequested.Inc()
  441. // If there is no binding or action, do not proceed. Leave default
  442. // log entry values (icon/title/id) and stop execution gracefully.
  443. if req.Binding == nil || req.Binding.Action == nil {
  444. log.Warnf("Action request has no binding/action; skipping execution")
  445. return false
  446. }
  447. req.logEntry.Binding = req.Binding
  448. req.logEntry.ActionConfigTitle = req.Binding.Action.Title
  449. req.logEntry.ActionTitle = entities.ParseTemplateWith(req.Binding.Action.Title, req.Binding.Entity)
  450. req.logEntry.ActionIcon = req.Binding.Action.Icon
  451. req.logEntry.ActionId = req.Binding.Action.ID
  452. req.logEntry.Tags = req.Tags
  453. req.executor.logmutex.Lock()
  454. if _, containsKey := req.executor.LogsByActionId[req.Binding.Action.ID]; !containsKey {
  455. req.executor.LogsByActionId[req.Binding.Action.ID] = make([]*InternalLogEntry, 0)
  456. }
  457. req.executor.LogsByActionId[req.Binding.Action.ID] = append(req.executor.LogsByActionId[req.Binding.Action.ID], req.logEntry)
  458. req.executor.logmutex.Unlock()
  459. log.WithFields(log.Fields{
  460. "actionTitle": req.logEntry.ActionTitle,
  461. "tags": req.Tags,
  462. }).Infof("Action requested")
  463. notifyListenersStarted(req)
  464. return true
  465. }
  466. func stepLogStart(req *ExecutionRequest) bool {
  467. log.WithFields(log.Fields{
  468. "actionTitle": req.logEntry.ActionTitle,
  469. "timeout": req.Binding.Action.Timeout,
  470. }).Infof("Action started")
  471. return true
  472. }
  473. func stepLogFinish(req *ExecutionRequest) bool {
  474. req.logEntry.ExecutionFinished = true
  475. log.WithFields(log.Fields{
  476. "actionTitle": req.logEntry.ActionTitle,
  477. "outputLength": len(req.logEntry.Output),
  478. "timedOut": req.logEntry.TimedOut,
  479. "exit": req.logEntry.ExitCode,
  480. }).Infof("Action finished")
  481. return true
  482. }
  483. func notifyListenersFinished(req *ExecutionRequest) {
  484. for _, listener := range req.executor.listeners {
  485. listener.OnExecutionFinished(req.logEntry)
  486. }
  487. }
  488. func notifyListenersStarted(req *ExecutionRequest) {
  489. for _, listener := range req.executor.listeners {
  490. listener.OnExecutionStarted(req.logEntry)
  491. }
  492. }
  493. func appendErrorToStderr(err error, logEntry *InternalLogEntry) {
  494. if err != nil {
  495. logEntry.Output = err.Error() + "\n\n" + logEntry.Output
  496. }
  497. }
  498. type OutputStreamer struct {
  499. Req *ExecutionRequest
  500. output bytes.Buffer
  501. }
  502. func (ost *OutputStreamer) Write(o []byte) (n int, err error) {
  503. for _, listener := range ost.Req.executor.listeners {
  504. listener.OnOutputChunk(o, ost.Req.TrackingID)
  505. }
  506. return ost.output.Write(o)
  507. }
  508. func (ost *OutputStreamer) String() string {
  509. return ost.output.String()
  510. }
  511. func buildEnv(args map[string]string) []string {
  512. ret := append(os.Environ(), "OLIVETIN=1")
  513. for k, v := range args {
  514. varName := fmt.Sprintf("%v", strings.TrimSpace(strings.ToUpper(k)))
  515. // Skip arguments that might not have a name (eg, confirmation), as this causes weird bugs on Windows.
  516. if varName == "" {
  517. continue
  518. }
  519. ret = append(ret, fmt.Sprintf("%v=%v", varName, v))
  520. }
  521. return ret
  522. }
  523. func stepExec(req *ExecutionRequest) bool {
  524. ctx, cancel := context.WithTimeout(context.Background(), time.Duration(req.Binding.Action.Timeout)*time.Second)
  525. defer cancel()
  526. streamer := &OutputStreamer{Req: req}
  527. cmd := buildCommand(ctx, req)
  528. if cmd == nil {
  529. req.logEntry.Output = "Cannot execute: no command arguments provided"
  530. log.Warn("Cannot execute: no command arguments provided")
  531. return false
  532. }
  533. prepareCommand(cmd, streamer, req)
  534. runerr := cmd.Start()
  535. req.logEntry.Process = cmd.Process
  536. waiterr := cmd.Wait()
  537. req.logEntry.ExitCode = int32(cmd.ProcessState.ExitCode())
  538. req.logEntry.Output = streamer.String()
  539. appendErrorToStderr(runerr, req.logEntry)
  540. appendErrorToStderr(waiterr, req.logEntry)
  541. if ctx.Err() == context.DeadlineExceeded {
  542. log.WithFields(log.Fields{
  543. "actionTitle": req.logEntry.ActionTitle,
  544. }).Warnf("Action timed out")
  545. // The context timeout should kill the process, but let's make sure.
  546. err := req.executor.Kill(req.logEntry)
  547. if err != nil {
  548. log.WithFields(log.Fields{
  549. "actionTitle": req.logEntry.ActionTitle,
  550. }).Warnf("could not kill process: %v", err)
  551. }
  552. req.logEntry.TimedOut = true
  553. req.logEntry.Output += "OliveTin::timeout - this action timed out after " + fmt.Sprintf("%v", req.Binding.Action.Timeout) + " seconds. If you need more time for this action, set a longer timeout. See https://docs.olivetin.app/action_customization/timeouts.html for more help."
  554. }
  555. req.logEntry.DatetimeFinished = time.Now()
  556. return true
  557. }
  558. func buildCommand(ctx context.Context, req *ExecutionRequest) *exec.Cmd {
  559. if req.useDirectExec {
  560. return wrapCommandDirect(ctx, req.execArgs)
  561. }
  562. return wrapCommandInShell(ctx, req.finalParsedCommand)
  563. }
  564. func prepareCommand(cmd *exec.Cmd, streamer *OutputStreamer, req *ExecutionRequest) {
  565. cmd.Stdout = streamer
  566. cmd.Stderr = streamer
  567. cmd.Env = buildEnv(req.Arguments)
  568. req.logEntry.ExecutionStarted = true
  569. }
  570. func stepExecAfter(req *ExecutionRequest) bool {
  571. if req.Binding.Action.ShellAfterCompleted == "" {
  572. return true
  573. }
  574. ctx, cancel := context.WithTimeout(context.Background(), time.Duration(req.Binding.Action.Timeout)*time.Second)
  575. defer cancel()
  576. var stdout bytes.Buffer
  577. var stderr bytes.Buffer
  578. args := map[string]string{
  579. "output": req.logEntry.Output,
  580. "exitCode": fmt.Sprintf("%v", req.logEntry.ExitCode),
  581. "ot_executionTrackingId": req.TrackingID,
  582. "ot_username": req.AuthenticatedUser.Username,
  583. }
  584. finalParsedCommand, err := parseCommandForReplacements(req.Binding.Action.ShellAfterCompleted, args, req.Binding.Entity)
  585. if err != nil {
  586. msg := "Could not prepare shellAfterCompleted command: " + err.Error() + "\n"
  587. req.logEntry.Output += msg
  588. log.Warn(msg)
  589. return true
  590. }
  591. cmd := wrapCommandInShell(ctx, finalParsedCommand)
  592. cmd.Stdout = &stdout
  593. cmd.Stderr = &stderr
  594. cmd.Env = buildEnv(args)
  595. runerr := cmd.Start()
  596. waiterr := cmd.Wait()
  597. req.logEntry.Output += "\n"
  598. req.logEntry.Output += "OliveTin::shellAfterCompleted stdout\n"
  599. req.logEntry.Output += stdout.String()
  600. req.logEntry.Output += "OliveTin::shellAfterCompleted stderr\n"
  601. req.logEntry.Output += stderr.String()
  602. req.logEntry.Output += "OliveTin::shellAfterCompleted errors and summary\n"
  603. appendErrorToStderr(runerr, req.logEntry)
  604. appendErrorToStderr(waiterr, req.logEntry)
  605. if ctx.Err() == context.DeadlineExceeded {
  606. req.logEntry.Output += "Your shellAfterCompleted command timed out."
  607. }
  608. req.logEntry.Output += fmt.Sprintf("Your shellAfterCompleted exited with code %v\n", cmd.ProcessState.ExitCode())
  609. req.logEntry.Output += "OliveTin::shellAfterCompleted output complete\n"
  610. return true
  611. }
  612. //gocyclo:ignore
  613. func stepTrigger(req *ExecutionRequest) bool {
  614. if req.Binding.Action.Triggers == nil {
  615. return true
  616. }
  617. if req.TriggerDepth >= MaxTriggerDepth {
  618. log.WithFields(log.Fields{
  619. "actionTitle": req.logEntry.ActionTitle,
  620. "depth": req.TriggerDepth,
  621. }).Warnf("Trigger action reached maximum depth of %v. Not triggering further actions.", MaxTriggerDepth)
  622. req.logEntry.Output += fmt.Sprintf("OliveTin::trigger - this action reached maximum trigger depth of %v. Not triggering further actions.", MaxTriggerDepth)
  623. return true
  624. }
  625. if len(req.Tags) > 0 && req.Tags[0] == "trigger" {
  626. log.Warnf("Trigger action is triggering another trigger action. This is allowed, but be careful not to create trigger loops.")
  627. }
  628. triggerLoop(req)
  629. return true
  630. }
  631. func triggerLoop(req *ExecutionRequest) {
  632. for _, triggerReq := range req.Binding.Action.Triggers {
  633. binding := req.executor.FindBindingByID(triggerReq)
  634. trigger := &ExecutionRequest{
  635. Binding: binding,
  636. TrackingID: uuid.NewString(),
  637. Tags: []string{"trigger"},
  638. AuthenticatedUser: req.AuthenticatedUser,
  639. Arguments: req.Arguments,
  640. Cfg: req.Cfg,
  641. TriggerDepth: req.TriggerDepth + 1,
  642. }
  643. req.executor.ExecRequest(trigger)
  644. }
  645. }
  646. func stepSaveLog(req *ExecutionRequest) bool {
  647. filename := fmt.Sprintf("%v.%v.%v", req.logEntry.ActionTitle, req.logEntry.DatetimeStarted.Unix(), req.logEntry.ExecutionTrackingID)
  648. saveLogResults(req, filename)
  649. saveLogOutput(req, filename)
  650. return true
  651. }
  652. func firstNonEmpty(one, two string) string {
  653. if one != "" {
  654. return one
  655. }
  656. return two
  657. }
  658. func saveLogResults(req *ExecutionRequest, filename string) {
  659. dir := firstNonEmpty(req.Binding.Action.SaveLogs.ResultsDirectory, req.Cfg.SaveLogs.ResultsDirectory)
  660. if dir != "" {
  661. data, err := yaml.Marshal(req.logEntry)
  662. if err != nil {
  663. log.Warnf("%v", err)
  664. }
  665. filepath := path.Join(dir, filename+".yaml")
  666. err = os.WriteFile(filepath, data, 0644)
  667. if err != nil {
  668. log.Warnf("%v", err)
  669. }
  670. }
  671. }
  672. func saveLogOutput(req *ExecutionRequest, filename string) {
  673. dir := firstNonEmpty(req.Binding.Action.SaveLogs.OutputDirectory, req.Cfg.SaveLogs.OutputDirectory)
  674. if dir != "" {
  675. data := req.logEntry.Output
  676. filepath := path.Join(dir, filename+".log")
  677. err := os.WriteFile(filepath, []byte(data), 0644)
  678. if err != nil {
  679. log.Warnf("%v", err)
  680. }
  681. }
  682. }