executor.go 25 KB

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