Initial commit

This repo is to supervise all processes in containers, in other
words inspect behaviors of dockers, and get the pid tree.

There are several ways for programs in user space to intereact with
kernel space:
- system calls, which can be found out in source path arch/x86/syscalls
- ioctl
- /proc virtual file system, to read kernel realtime info
- nerlink socket

the pid we should pay attention to is /usr/bin/containerd, which may
come from service docker-daemon and ppid is 1. Each time a docker is
start or stop, this forks a pid, the pid then forks, that's the main
process of the docker.

To grub the info of pid create or exit, this program is based on
go-libauditd, which uses netlink socket to hear from kernel about
audit log. What's worrying is that one event is always devided into
several entries, and several events may be received alternately.

So, from my point of view, which program has 3 coroutines and 2
channels. the first receives raw event message from audit, then
throw it to channel 1; the second listen to channel 1, and organizes
each event until there's a EOE, then throw to channel 2; the third
discover event from channel 2, deal with th event, such as create or
delete pid. Specially, since two relative infomation(pid 1 fork pid2,
then pid 1 exits)may comes out of order, deletion mast be delayed for
some time(may 1 second), to keep the process tree correct.

2 files changed, 345 insertions, 0 deletions

diff --git a/old/audit.go b/old/audit.go
new file mode 100644
index 0000000..2b9faa5
--- /dev/null
+++ b/old/audit.go
@@ -0,0 +1,127 @@
+package main
+
+import (
+        "errors"
+        "flag"
+        "fmt"
+        "io"
+        "log"
+        "os"
+
+        "github.com/elastic/go-libaudit/v2"
+        "github.com/elastic/go-libaudit/v2/auparse"
+)
+
+var (
+        fs          = flag.NewFlagSet("audit", flag.ExitOnError)
+        diag        = fs.String("diag", "", "dump raw information from kernel to file")
+        rate        = fs.Uint("rate", 0, "rate limit in kernel (default 0, no rate limit)")
+        backlog     = fs.Uint("backlog", 8192, "backlog limit")
+        immutable   = fs.Bool("immutable", false, "make kernel audit settings immutable (requires reboot to undo)")
+        receiveOnly = fs.Bool("ro", false, "receive only using multicast, requires kernel 3.16+")
+)
+
+func main() {
+        if err := fs.Parse(os.Args[1:]); err != nil {
+                log.Fatal(err)
+        }
+
+        if err := read(); err != nil {
+                log.Fatalf("error: %v", err)
+        }
+}
+
+func read() error {
+        if os.Geteuid() != 0 {
+                return errors.New("you must be root to receive audit data")
+        }
+
+        // Write netlink response to a file for further analysis or for writing
+        // tests cases.
+        var diagWriter io.Writer
+        if *diag != "" {
+                f, err := os.OpenFile(*diag, os.O_CREATE|os.O_RDWR|os.O_TRUNC, 0o600)
+                if err != nil {
+                        return err
+                }
+                defer f.Close()
+                diagWriter = f
+        }
+
+        log.Println("starting netlink client")
+
+        var err error
+        var client *libaudit.AuditClient
+        if *receiveOnly {
+                client, err = libaudit.NewMulticastAuditClient(diagWriter)
+                if err != nil {
+                        return fmt.Errorf("failed to create receive-only audit client: %w", err)
+                }
+                defer client.Close()
+        } else {
+                client, err = libaudit.NewAuditClient(diagWriter)
+                if err != nil {
+                        return fmt.Errorf("failed to create audit client: %w", err)
+                }
+                defer client.Close()
+
+                status, err := client.GetStatus()
+                if err != nil {
+                        return fmt.Errorf("failed to get audit status: %w", err)
+                }
+                log.Printf("received audit status=%+v", status)
+
+                if status.Enabled == 0 {
+                        log.Println("enabling auditing in the kernel")
+                        if err = client.SetEnabled(true, libaudit.WaitForReply); err != nil {
+                                return fmt.Errorf("failed to set enabled=true: %w", err)
+                        }
+                }
+
+                if status.RateLimit != uint32(*rate) {
+                        log.Printf("setting rate limit in kernel to %v", *rate)
+                        if err = client.SetRateLimit(uint32(*rate), libaudit.NoWait); err != nil {
+                                return fmt.Errorf("failed to set rate limit to unlimited: %w", err)
+                        }
+                }
+
+                if status.BacklogLimit != uint32(*backlog) {
+                        log.Printf("setting backlog limit in kernel to %v", *backlog)
+                        if err = client.SetBacklogLimit(uint32(*backlog), libaudit.NoWait); err != nil {
+                                return fmt.Errorf("failed to set backlog limit: %w", err)
+                        }
+                }
+
+                if status.Enabled != 2 && *immutable {
+                        log.Printf("setting kernel settings as immutable")
+                        if err = client.SetImmutable(libaudit.NoWait); err != nil {
+                                return fmt.Errorf("failed to set kernel as immutable: %w", err)
+                        }
+                }
+
+                log.Printf("sending message to kernel registering our PID (%v) as the audit daemon", os.Getpid())
+                if err = client.SetPID(libaudit.NoWait); err != nil {
+                        return fmt.Errorf("failed to set audit PID: %w", err)
+                }
+        }
+
+        return receive(client)
+}
+
+func receive(r *libaudit.AuditClient) error {
+        for {
+                rawEvent, err := r.Receive(false)
+                if err != nil {
+                        return fmt.Errorf("receive failed: %w", err)
+                }
+
+                // Messages from 1300-2999 are valid audit messages.
+                if rawEvent.Type < auparse.AUDIT_USER_AUTH ||
+                        rawEvent.Type > auparse.AUDIT_LAST_USER_MSG2 {
+                        continue
+                }
+
+                fmt.Printf("type=%v msg=%s\n", rawEvent.Type, rawEvent.Data)
+                // fmt.Printf("type=%v\n", rawEvent.Type)
+        }
+}
diff --git a/old/auparse.go b/old/auparse.go
new file mode 100644
index 0000000..53b0c92
--- /dev/null
+++ b/old/auparse.go
@@ -0,0 +1,218 @@
+package main
+
+import (
+        "bufio"
+        "encoding/json"
+        "flag"
+        "fmt"
+        "io"
+        "log"
+        "os"
+        "time"
+
+        "gopkg.in/yaml.v3"
+
+        "github.com/elastic/go-libaudit/v2"
+        "github.com/elastic/go-libaudit/v2/aucoalesce"
+        "github.com/elastic/go-libaudit/v2/auparse"
+)
+
+var (
+        fs        = flag.NewFlagSet("auparse", flag.ExitOnError)
+        in        = fs.String("in", "-", "input file (defaults to stdin)")
+        out       = fs.String("out", "-", "output file (defaults to stdout)")
+        interpret = fs.Bool("i", false, "interpret and normalize messages")
+        idLookup  = fs.Bool("id", true, "lookup uid and gid values in messages (requires -i)")
+        format    = fs.String("format", "", "output format, possible values - json, yaml, text (default)")
+)
+
+func main() {
+        if err := fs.Parse(os.Args[1:]); err != nil {
+                log.Fatal(err)
+        }
+
+        if err := processLogs(); err != nil {
+                log.Fatalf("error: %v", err)
+        }
+}
+
+func input() (io.ReadCloser, error) {
+        if *in == "-" {
+                return os.Stdin, nil
+        }
+
+        return os.Open(*in)
+}
+
+func output() (io.WriteCloser, error) {
+        if *out == "-" {
+                return os.Stdout, nil
+        }
+
+        return os.OpenFile(*out, os.O_CREATE|os.O_RDWR|os.O_TRUNC, 0o600)
+}
+
+func processLogs() error {
+        input, err := input()
+        if err != nil {
+                return err
+        }
+        defer input.Close()
+
+        output, err := output()
+        if err != nil {
+                return err
+        }
+        defer output.Close()
+
+        reassembler, err := libaudit.NewReassembler(5, 2*time.Second, &streamHandler{output})
+        if err != nil {
+                return fmt.Errorf("failed to create reassmbler: %w", err)
+        }
+        defer reassembler.Close()
+
+        // Start goroutine to periodically purge timed-out events.
+        go func() {
+                t := time.NewTicker(500 * time.Millisecond)
+                defer t.Stop()
+                for range t.C {
+                        if reassembler.Maintain() != nil {
+                                return
+                        }
+                }
+        }()
+
+        // Process lines from the input.
+        s := bufio.NewScanner(input)
+        for s.Scan() {
+                line := s.Text()
+
+                auditMsg, err := auparse.ParseLogLine(line)
+                if err != nil {
+                        log.Printf("failed to parse message header: %v", err)
+                }
+
+                reassembler.PushMessage(auditMsg)
+        }
+
+        return nil
+}
+
+type streamHandler struct {
+        output io.Writer
+}
+
+func (s *streamHandler) ReassemblyComplete(msgs []*auparse.AuditMessage) {
+        if err := s.outputMultipleMessages(msgs); err != nil {
+                log.Printf("[WARN] failed writing message to output: %v", err)
+        }
+}
+
+func (*streamHandler) EventsLost(count int) {
+        log.Printf("detected the loss of %v sequences.", count)
+}
+
+func (s *streamHandler) outputMultipleMessages(msgs []*auparse.AuditMessage) error {
+        var err error
+        if !*interpret {
+                if _, err = s.output.Write([]byte("---\n")); err != nil {
+                        return err
+                }
+                for _, m := range msgs {
+                        if err = s.outputSingleMessage(m); err != nil {
+                                return err
+                        }
+                }
+                return nil
+        }
+
+        event, err := aucoalesce.CoalesceMessages(msgs)
+        if err != nil {
+                log.Printf("failed to coalesce messages: %v", err)
+                return nil
+        }
+
+        if *idLookup {
+                aucoalesce.ResolveIDs(event)
+        }
+
+        switch *format {
+        case "json":
+                if err := s.printJSON(event); err != nil {
+                        log.Printf("failed to marshal event to JSON: %v", err)
+                }
+        case "yaml":
+                if _, err := s.output.Write([]byte("---\n")); err != nil {
+                        return err
+                }
+                if err := s.printYAML(event); err != nil {
+                        log.Printf("failed to marshal message to YAML: %v", err)
+                }
+        default:
+                sm := event.Summary
+                if _, err := s.output.Write([]byte("---\n")); err != nil {
+                        return err
+                }
+
+                _, err := fmt.Fprintf(
+                        s.output,
+                        `time="%v" sequence=%v category=%v type=%v actor=%v/%v action=%v thing=%v/%v how=%v tags=%v`+"\n",
+                        event.Timestamp, event.Sequence, event.Category, event.Type, sm.Actor.Primary, sm.Actor.Secondary,
+                        sm.Action, sm.Object.Primary, sm.Object.Secondary, sm.How, event.Tags,
+                )
+                if err != nil {
+                        return err
+                }
+        }
+        return nil
+}
+
+func (s *streamHandler) outputSingleMessage(m *auparse.AuditMessage) error {
+        switch *format {
+        case "json":
+                if err := s.printJSON(m.ToMapStr()); err != nil {
+                        log.Printf("failed to marshal message to JSON: %v", err)
+                }
+        case "yaml":
+                if err := s.printYAML(m.ToMapStr()); err != nil {
+                        log.Printf("failed to marshal message to YAML: %v", err)
+                }
+        default:
+                if _, err := fmt.Fprintf(
+                        s.output,
+                        "type=%v msg=%v\n",
+                        m.RecordType, m.RawData,
+                ); err != nil {
+                        return err
+                }
+        }
+        return nil
+}
+
+func (s *streamHandler) printJSON(v interface{}) error {
+        jsonBytes, err := json.Marshal(v)
+        if err != nil {
+                return err
+        }
+        if _, err = s.output.Write(jsonBytes); err != nil {
+                return err
+        }
+        if _, err = s.output.Write([]byte("\n")); err != nil {
+                return err
+        }
+        return nil
+}
+
+func (s *streamHandler) printYAML(v interface{}) error {
+        yamlBytes, err := yaml.Marshal(v)
+        if err != nil {
+                return err
+        }
+        if _, err = s.output.Write(yamlBytes); err != nil {
+                return err
+        }
+        if _, err = s.output.Write([]byte("\n")); err != nil {
+                return err
+        }
+        return nil
+}