main.go

// +build linux

package main

import (
    "os"
    "flag"
    "fmt"
    "time"
    "sync"
    "runtime"
    "strconv"
    "strings"
    "syscall"
    "encoding/json"
    "github.com/NixM0nk3y/go-audit-daemon/common"
    "github.com/pkg/errors"
    "github.com/golang/glog"
    "github.com/elastic/go-libaudit"
    "github.com/elastic/go-libaudit/aucoalesce"
    "github.com/elastic/go-libaudit/auparse"
    "github.com/elastic/go-libaudit/rule"
)

const (
        auditLocked = 2

        unicast   = "unicast"
        multicast = "multicast"

        lostEventsUpdateInterval        = time.Second * 15
        maxDefaultStreamBufferConsumers = 4

        bsKernel backpressureStrategy = 1 << iota
        bsUserSpace
        bsAuto
)

// Event contains the data generated by a MetricSet.
type Event struct {
	RootFields      common.MapStr // Fields that will be added to the root of the event.
	ModuleFields    common.MapStr // Fields that will be namespaced under [module].
	MetricSetFields common.MapStr // Fields that will be namespaced under [module].[metricset].

	Index     string        // Index name prefix. If set overwrites the default prefix.
	Namespace string        // Fully qualified namespace to use for MetricSetFields.
	Timestamp time.Time     // Timestamp when the event data was collected.
	Error     error         // Error that occurred while collecting the event data.
	Host      string        // Host from which the data was collected.
	Took      time.Duration // Amount of time it took to collect the event data.
}

type backpressureStrategy uint8

type AuditSet struct {
        config     Config
        client     *libaudit.AuditClient
        encoder    *json.Encoder
        kernelLost struct {
                enabled bool
                counter uint32
        }
        backpressureStrategy backpressureStrategy
}

// stream type

// stream receives callbacks from the libaudit.Reassembler for completed events
// or lost events that are detected by gaps in sequence numbers.
type stream struct {
        done <-chan struct{}
        out  chan<- []*auparse.AuditMessage
}

func (s *stream) ReassemblyComplete(msgs []*auparse.AuditMessage) {
        select {
        case <-s.done:
                return
        case s.out <- msgs:
        }
}

func (s *stream) EventsLost(count int) {
}


// nonBlockingStream behaves as stream above, except that it will never block
// on backpressure from the publishing pipeline.
// Instead, events will be discarded.
type nonBlockingStream stream

func (s *nonBlockingStream) ReassemblyComplete(msgs []*auparse.AuditMessage) {
        select {
        case <-s.done:
                return
        case s.out <- msgs:
        default:
        }
}

func (s *nonBlockingStream) EventsLost(count int) {
        (*stream)(s).EventsLost(count)
}

func usage() {
	fmt.Fprintf(os.Stderr, "usage: example -stderrthreshold=[INFO|WARN|FATAL] -log_dir=[string]\n", )
	flag.PrintDefaults()
	os.Exit(2)
}

func init() {
	flag.Usage = usage
	flag.Parse()
}

func main() {

    config := defaultConfig

    _, _, kernel, _ := kernelVersion()

    glog.Infof("auditd module is running as euid=%v on kernel=%v", os.Geteuid(), kernel)

    client, err := newAuditClient(&config)

    if err != nil {
        glog.Error("Failed to create new audit client")
	os.Exit(2)
    }

    defer client.Close()

    if err := config.loadRules(); err != nil {
        glog.Errorln("Failure load audit rules", "error", err)
        return
    }

    client.DeleteRules() 

    if err := addRules(config); err != nil {
        glog.Errorln("Failure adding audit rules", "error", err)
        return
    }

    as := &AuditSet{
        client: client,
        config: config,
        encoder: json.NewEncoder(os.Stdout),
        backpressureStrategy: getBackpressureStrategy(config.BackpressureStrategy),
    }

    status, err := client.GetStatus()
    if err != nil {
        glog.Error("failed to get audit status")
        os.Exit(2)
    }

    glog.Infof("received audit status=%+v", status)

    done := make(chan struct{})
    defer close(done)

    out, err := receiveEvents(as, done)
    if err != nil {
        glog.Errorln("Failure receiving audit events", "error", err)

        return
    }

    if as.kernelLost.enabled {

        client, err := libaudit.NewAuditClient(nil)
        if err != nil {
            glog.Errorln("Failure creating audit monitoring client", "error", err)
        }
        go func() {
            defer client.Close()
            timer := time.NewTicker(lostEventsUpdateInterval)
            defer timer.Stop()
            for {
                select {
                    case <-done:
                        return
                    case <-timer.C:
                        if status, err := client.GetStatus(); err == nil {
                            updateKernelLostMetric(as, status.Lost)
                        } else {
                            glog.Error("get status request failed:", err)
                        }
                }
            }
        }()
    }

    // Spawn the stream buffer consumers
    numConsumers := as.config.StreamBufferConsumers

    // By default (stream_buffer_consumers=0) use as many consumers as local CPUs
    // with a max of `maxDefaultStreamBufferConsumers`
    if numConsumers == 0 {
        if numConsumers = runtime.GOMAXPROCS(-1); numConsumers > maxDefaultStreamBufferConsumers {
            numConsumers = maxDefaultStreamBufferConsumers
        }
    }
    var wg sync.WaitGroup
    wg.Add(numConsumers)

    for i := 0; i < numConsumers; i++ {
        go func() {
            defer wg.Done()
            for {
                select {
                    case <-done:
                        return
                    case msgs := <-out:
                        out := buildEvent(msgs, as.config)

                        err = as.encoder.Encode(out)

                        if err != nil {
                            glog.Errorln("Failure to serialise event","error",err)

                        }
                }
            }
        }()
    }
    wg.Wait()

    return
}

func addRules(config Config) error {

        rules := config.rules()

        if len(rules) == 0 {
                glog.Info("No audit_rules were specified.")
                return nil
        }

        client, err := libaudit.NewAuditClient(nil)

        if err != nil {
                return errors.Wrap(err, "failed to create audit client for adding rules")
        }
        defer client.Close()

        // Don't attempt to change configuration if audit rules are locked (enabled == 2).
        // Will result in EPERM.
        status, err := client.GetStatus()
        if err != nil {
                err = errors.Wrap(err, "failed to get audit status before adding rules")
                return err
        }
        if status.Enabled == auditLocked {
                return errors.New("Skipping rule configuration: Audit rules are locked")
        }

        // Delete existing rules.
        n, err := client.DeleteRules()
        if err != nil {
                return errors.Wrap(err, "failed to delete existing rules")
        }
        glog.Infof("Deleted %v pre-existing audit rules.", n)

        // Add rule to ignore syscalls from this process
        if rule, err := buildPIDIgnoreRule(os.Getpid()); err == nil {
                rules = append([]auditRule{rule}, rules...)
        } else {
                glog.Errorf("Failed to build a rule to ignore self: %v", err)
        }
        // Add rules from config.
        var failCount int
        for _, rule := range rules {
                if err = client.AddRule(rule.data); err != nil {
                        // Treat rule add errors as warnings and continue.
                        err = errors.Wrapf(err, "failed to add audit rule '%v'", rule.flags)
                        glog.Warningln("Failure adding audit rule", "error", err)
                        failCount++
                }
        }
        glog.Infof("Successfully added %d of %d audit rules.",
                len(rules)-failCount, len(rules))
        return nil
}

func buildEvent(msgs []*auparse.AuditMessage, config Config) Event {

    auditEvent, err := aucoalesce.CoalesceMessages(msgs)

        if err != nil {
                out := Event{MetricSetFields: common.MapStr{}}
                addMessages(msgs, out.MetricSetFields)
                return out
        }

        if config.ResolveIDs {
                aucoalesce.ResolveIDs(auditEvent)
        }

        out := Event{
                Timestamp: auditEvent.Timestamp,
                RootFields: common.MapStr{
                        "event": common.MapStr{
                                "category": auditEvent.Category.String(),
                                "type":     strings.ToLower(auditEvent.Type.String()),
                                "action":   auditEvent.Summary.Action,
                        },
                },
                ModuleFields: common.MapStr{
                        "sequence": auditEvent.Sequence,
                        "result":   auditEvent.Result,
                        "session":  auditEvent.Session,
                        "data":     createAuditdData(auditEvent.Data),
                },
        }

        // Add root level fields.
        addUser(auditEvent.User, out.RootFields)
        addProcess(auditEvent.Process, out.RootFields)
        addFile(auditEvent.File, out.RootFields)
        addAddress(auditEvent.Source, "source", out.RootFields)
        addAddress(auditEvent.Dest, "destination", out.RootFields)
        addNetwork(auditEvent.Net, out.RootFields)
        if len(auditEvent.Tags) > 0 {
                out.RootFields.Put("tags", auditEvent.Tags)
        }

        // Add module fields.
        m := out.ModuleFields
        if auditEvent.Summary.Actor.Primary != "" {
                m.Put("summary.actor.primary", auditEvent.Summary.Actor.Primary)
        }
        if auditEvent.Summary.Actor.Secondary != "" {
                m.Put("summary.actor.secondary", auditEvent.Summary.Actor.Secondary)
        }
        if auditEvent.Summary.Object.Primary != "" {
                m.Put("summary.object.primary", auditEvent.Summary.Object.Primary)
        }
        if auditEvent.Summary.Object.Secondary != "" {
                m.Put("summary.object.secondary", auditEvent.Summary.Object.Secondary)
        }
        if auditEvent.Summary.Object.Type != "" {
                m.Put("summary.object.type", auditEvent.Summary.Object.Type)
        }
        if auditEvent.Summary.How != "" {
                m.Put("summary.how", auditEvent.Summary.How)
        }
        if len(auditEvent.Paths) > 0 {
                m.Put("paths", auditEvent.Paths)
        }
        if config.Warnings && len(auditEvent.Warnings) > 0 {
                warnings := make([]string, 0, len(auditEvent.Warnings))
                for _, err := range auditEvent.Warnings {
                        warnings = append(warnings, err.Error())
                }
                m.Put("warnings", warnings)
                addMessages(msgs, m)
        }
        if config.RawMessage {
                addMessages(msgs, m)
        }

        return out
}

func receiveEvents(as *AuditSet, done <-chan struct{}) (<-chan []*auparse.AuditMessage, error) {

    if err := initClient(as); err != nil {
        return nil, err
    }

    out := make(chan []*auparse.AuditMessage, as.config.StreamBufferQueueSize)

    var st libaudit.Stream = &stream{done, out}
    if as.backpressureStrategy&bsUserSpace != 0 {
        // "user-space" backpressure mitigation strategy
        //
        // Consume events from our side as fast as possible, by dropping events
        // if the publishing pipeline would block.
        glog.Info("Using non-blocking stream to prevent backpressure propagating to the kernel.")
        st = &nonBlockingStream{done, out}
    }

    reassembler, err := libaudit.NewReassembler(int(as.config.ReassemblerMaxInFlight), as.config.ReassemblerTimeout, st)
    if err != nil {
        return nil, errors.Wrap(err, "failed to create Reassembler")
    }
    go maintain(done, reassembler)

    go func() {
        defer glog.Info("receiveEvents goroutine exited")
        defer close(out)
        defer reassembler.Close()

        for {
            raw, err := as.client.Receive(false)
            if err != nil {
                if errors.Cause(err) == syscall.EBADF {
                    // Client has been closed.
                    break
                }
                continue
            }

            if filterRecordType(raw.Type) {
                continue
            }
            if err := reassembler.Push(raw.Type, raw.Data); err != nil {
                glog.Infoln("Dropping audit message",
                            "record_type", raw.Type,
                            "message", string(raw.Data),
                            "error", err)
                continue
            }
        }
    }()


    return out, nil
}

// maintain periodically evicts timed-out events from the Reassembler. This
// function will block until the done channel is closed or the Reassembler is
// closed.
func maintain(done <-chan struct{}, reassembler *libaudit.Reassembler) {
    tick := time.NewTicker(500 * time.Millisecond)
    defer tick.Stop()

    for {
        select {
            case <-done:
                return
            case <-tick.C:
                if err := reassembler.Maintain(); err != nil {
                    return
                }
        }
    }
}


func initClient(as *AuditSet) error {

        if as.config.SocketType == "multicast" {
                // This request will fail with EPERM if this process does not have
                // CAP_AUDIT_CONTROL, but we will ignore the response. The user will be
                // required to ensure that auditing is enabled if the process is only
                // given CAP_AUDIT_READ.
        }

        // Unicast client initialization (requires CAP_AUDIT_CONTROL and that the
        // process be in initial PID namespace).
        status, err := as.client.GetStatus()
        if err != nil {
                return errors.Wrap(err, "failed to get audit status")
        }
        as.kernelLost.enabled = true
        as.kernelLost.counter = status.Lost

        glog.Infoln("audit status from kernel at start", "audit_status", status)

        if status.Enabled == auditLocked {
                return errors.New("failed to configure: The audit system is locked")
        }

        if fm, _ := as.config.failureMode(); status.Failure != fm {
                if err = as.client.SetFailure(libaudit.FailureMode(fm), libaudit.NoWait); err != nil {
                        return errors.Wrap(err, "failed to set audit failure mode in kernel")
                }
        }

        if status.BacklogLimit != as.config.BacklogLimit {
                if err = as.client.SetBacklogLimit(as.config.BacklogLimit, libaudit.NoWait); err != nil {
                        return errors.Wrap(err, "failed to set audit backlog limit in kernel")
                }
        }

        if as.backpressureStrategy&(bsKernel|bsAuto) != 0 {
                // "kernel" backpressure mitigation strategy
                //
                // configure the kernel to drop audit events immediately if the
                // backlog queue is full.
                if status.FeatureBitmap&libaudit.AuditFeatureBitmapBacklogWaitTime != 0 {
                        glog.Info("Setting kernel backlog wait time to prevent backpressure propagating to the kernel.")
                        if err = as.client.SetBacklogWaitTime(0, libaudit.NoWait); err != nil {
                                return errors.Wrap(err, "failed to set audit backlog wait time in kernel")
                        }
                } else {
                        if as.backpressureStrategy == bsAuto {
                                glog.Warning("setting backlog wait time is not supported in this kernel. Enabling workaround.")
                                as.backpressureStrategy |= bsUserSpace
                        } else {
                                return errors.New("kernel backlog wait time not supported by kernel, but required by backpressure_strategy")
                        }
                }
        }

        if as.backpressureStrategy&(bsKernel|bsUserSpace) == bsUserSpace && as.config.RateLimit == 0 {
                // force a rate limit if the user-space strategy will be used without
                // corresponding backlog_wait_time setting in the kernel
                as.config.RateLimit = 5000
        }

        if status.RateLimit != as.config.RateLimit {
                if err = as.client.SetRateLimit(as.config.RateLimit, libaudit.NoWait); err != nil {
                        return errors.Wrap(err, "failed to set audit rate limit in kernel")
                }
        }

        if status.Enabled == 0 {
                if err = as.client.SetEnabled(true, libaudit.NoWait); err != nil {
                        return errors.Wrap(err, "failed to enable auditing in the kernel")
                }
        }
        if err := as.client.WaitForPendingACKs(); err != nil {
                return errors.Wrap(err, "failed to wait for ACKs")
        }
        if err := as.client.SetPID(libaudit.WaitForReply); err != nil {
                if errno, ok := err.(syscall.Errno); ok && errno == syscall.EEXIST && status.PID != 0 {
                        return fmt.Errorf("failed to set audit PID. An audit process is already running (PID %d)", status.PID)
                }
                return errors.Wrapf(err, "failed to set audit PID (current audit PID %d)", status.PID)
        }
        return nil
}

func newAuditClient(c *Config) (*libaudit.AuditClient, error) {
        var err error
        c.SocketType, err = determineSocketType(c)
        if err != nil {
                return nil, err
        }
        glog.Infof("socket_type=%s will be used.", c.SocketType)

        if c.SocketType == multicast {
                return libaudit.NewMulticastAuditClient(nil)
        }
        return libaudit.NewAuditClient(nil)
}

func kernelVersion() (major, minor int, full string, err error) {
        var uname syscall.Utsname
        if err := syscall.Uname(&uname); err != nil {
                return 0, 0, "", err
        }

        length := len(uname.Release)
        data := make([]byte, length)
        for i, v := range uname.Release {
                if v == 0 {
                        length = i
                        break
                }
                data[i] = byte(v)
        }

        release := string(data[:length])
        parts := strings.SplitN(release, ".", 3)
        if len(parts) < 2 {
                return 0, 0, release, errors.Errorf("failed to parse uname release '%v'", release)
        }

        major, err = strconv.Atoi(parts[0])
        if err != nil {
                return 0, 0, release, errors.Wrapf(err, "failed to parse major version from '%v'", release)
        }

        minor, err = strconv.Atoi(parts[1])
        if err != nil {
                return 0, 0, release, errors.Wrapf(err, "failed to parse minor version from '%v'", release)
        }

        return major, minor, release, nil
}

func determineSocketType(c *Config) (string, error) {
        client, err := libaudit.NewAuditClient(nil)
        if err != nil {
                if c.SocketType == "" {
                        return "", errors.Wrap(err, "failed to create audit client")
                }
                // Ignore errors if a socket type has been specified. It will fail during
                // further setup and its necessary for unit tests to pass
                return c.SocketType, nil
        }
        defer client.Close()
        status, err := client.GetStatus()
        if err != nil {
                if c.SocketType == "" {
                        return "", errors.Wrap(err, "failed to get audit status")
                }
                return c.SocketType, nil
        }
        rules := c.rules()

        isLocked := status.Enabled == auditLocked
        hasMulticast := hasMulticastSupport()
        hasRules := len(rules) > 0

        const useAutodetect = "Remove the socket_type option to have auditbeat " +
                "select the most suitable subscription method."
        switch c.SocketType {
        case unicast:
                if isLocked {
                        glog.Errorf("requested unicast socket_type is not available "+
                                "because audit configuration is locked in the kernel "+
                                "(enabled=2). %s", useAutodetect)
                        return "", errors.New("unicast socket_type not available")
                }
                return c.SocketType, nil

        case multicast:
                if hasMulticast {
                        if hasRules {
                                glog.Warning("The audit rules specified in the configuration " +
                                        "cannot be applied when using a multicast socket_type.")
                        }
                        return c.SocketType, nil
                }
                glog.Errorf("socket_type is set to multicast but based on the "+
                        "kernel version, multicast audit subscriptions are not supported. %s",
                        useAutodetect)
                return "", errors.New("multicast socket_type not available")

        default:
                // attempt to determine the optimal socket_type
                if hasMulticast {
                        if hasRules {
                                if isLocked {
                                        glog.Warning("Audit rules specified in the configuration " +
                                                "cannot be applied because the audit rules have been locked " +
                                                "in the kernel (enabled=2). A multicast audit subscription " +
                                                "will be used instead, which does not support setting rules")
                                        return multicast, nil
                                }
                                return unicast, nil
                        }
                        return multicast, nil
                }
                if isLocked {
                        glog.Errorf("Cannot continue: audit configuration is locked " +
                                "in the kernel (enabled=2) which prevents using unicast " +
                                "sockets. Multicast audit subscriptions are not available " +
                                "in this kernel. Disable locking the audit configuration " +
                                "to use auditbeat.")
                        return "", errors.New("no connection to audit available")
                }
                return unicast, nil
        }

}

func updateKernelLostMetric(as *AuditSet, lost uint32) {
        if !as.kernelLost.enabled {
                return
        }
        delta := int64(lost - as.kernelLost.counter)
        if delta >= 0 {
                glog.Warningf("kernel lost events: %d (total: %d)", delta, lost)
        } else {
                glog.Warningf("kernel lost event counter reset from %d to %d", as.kernelLost, lost)
        }
        as.kernelLost.counter = lost
}

func getBackpressureStrategy(value string) backpressureStrategy {
        switch value {
        case "kernel":
                return bsKernel
        case "userspace", "user-space":
                return bsUserSpace
        case "auto":
                return bsAuto
        case "both":
                return bsKernel | bsUserSpace
        case "none":
                return 0
        default:
                glog.Warning("Unknown value for the 'backpressure_strategy' option. Using default.")
                fallthrough
        case "", "default":
                return bsAuto
        }
}

func filterRecordType(typ auparse.AuditMessageType) bool {
        switch {
        // REPLACE messages are tests to check if Auditbeat is still healthy by
        // seeing if unicast messages can be sent without error from the kernel.
        // Ignore them.
        case typ == auparse.AUDIT_REPLACE:
                return true
        // Messages from 1300-2999 are valid audit message types.
        case typ < auparse.AUDIT_USER_AUTH || typ > auparse.AUDIT_LAST_USER_MSG2:
                return true
        }

        return false
}

func hasMulticastSupport() bool {
        // Check the kernel version because 3.16+ should have multicast
        // support.
        major, minor, _, err := kernelVersion()
        if err != nil {
                // Assume not supported.
                return false
        }

        switch {
        case major > 3,
                major == 3 && minor >= 16:
                return true
        }

        return false
}

func addUser(u aucoalesce.User, m common.MapStr) {
        user := make(common.MapStr, len(u.IDs))
        m.Put("user", user)

        for id, value := range u.IDs {
                user[id] = value
                if len(u.SELinux) > 0 {
                        user["selinux"] = u.SELinux
                }
                if len(u.Names) > 0 {
                        user["name_map"] = u.Names
                }
        }
}

func addProcess(p aucoalesce.Process, m common.MapStr) {
        if p.IsEmpty() {
                return
        }

        process := common.MapStr{}
        m.Put("process", process)
        if p.PID != "" {
                process["pid"] = p.PID
        }
        if p.PPID != "" {
                process["ppid"] = p.PPID
        }
        if p.Title != "" {
                process["title"] = p.Title
        }
        if p.Name != "" {
                process["name"] = p.Name
        }
        if p.Exe != "" {
                process["exe"] = p.Exe
        }
        if p.CWD != "" {
                process["cwd"] = p.CWD
        }
        if len(p.Args) > 0 {
                process["args"] = p.Args
        }
}

func addFile(f *aucoalesce.File, m common.MapStr) {
        if f == nil {
                return
        }

        file := common.MapStr{}
        m.Put("file", file)
        if f.Path != "" {
                file["path"] = f.Path
        }
        if f.Device != "" {
                file["device"] = f.Device
        }
        if f.Inode != "" {
                file["inode"] = f.Inode
        }
        if f.Mode != "" {
                file["mode"] = f.Mode
        }
        if f.UID != "" {
                file["uid"] = f.UID
        }
        if f.GID != "" {
                file["gid"] = f.GID
        }
        if f.Owner != "" {
                file["owner"] = f.Owner
        }
        if f.Group != "" {
                file["group"] = f.Group
        }
        if len(f.SELinux) > 0 {
                file["selinux"] = f.SELinux
        }
}

func addAddress(addr *aucoalesce.Address, key string, m common.MapStr) {
        if addr == nil {
                return
        }

        address := common.MapStr{}
        m.Put(key, address)
        if addr.Hostname != "" {
                address["hostname"] = addr.Hostname
        }
        if addr.IP != "" {
                address["ip"] = addr.IP
        }
        if addr.Port != "" {
                address["port"] = addr.Port
        }
        if addr.Path != "" {
                address["path"] = addr.Path
        }
}

func addNetwork(net *aucoalesce.Network, m common.MapStr) {
        if net == nil {
                return
        }

        network := common.MapStr{
                "direction": net.Direction,
        }
        m.Put("network", network)
}

func addMessages(msgs []*auparse.AuditMessage, m common.MapStr) {
        _, added := m["messages"]
        if !added && len(msgs) > 0 {
                rawMsgs := make([]string, 0, len(msgs))
                for _, msg := range msgs {
                        rawMsgs = append(rawMsgs, "type="+msg.RecordType.String()+" msg="+msg.RawData)
                }
                m["messages"] = rawMsgs
        }
}

func createAuditdData(data map[string]string) common.MapStr {
        out := make(common.MapStr, len(data))
        for key, v := range data {
                if strings.HasPrefix(key, "socket_") {
                        out.Put("socket."+key[7:], v)
                        continue
                }

                out.Put(key, v)
        }
        return out
}

func buildPIDIgnoreRule(pid int) (ruleData auditRule, err error) {
        r := rule.SyscallRule{
                Type:   rule.AppendSyscallRuleType,
                List:   "exit",
                Action: "never",
                Filters: []rule.FilterSpec{
                        {
                                Type:       rule.ValueFilterType,
                                LHS:        "pid",
                                Comparator: "=",
                                RHS:        strconv.Itoa(pid),
                        },
                },
                Syscalls: []string{"all"},
                Keys:     nil,
        }
        ruleData.flags = fmt.Sprintf("-A exit,never -F pid=%d -S all", pid)
        ruleData.data, err = rule.Build(&r)
        return ruleData, err
}