package importer import ( "context" "errors" "net" nethttp "net/http" "sync/atomic" "time" "github.com/activecm/rita/v5/config" "github.com/activecm/rita/v5/database" "github.com/activecm/rita/v5/importer/zeektypes" zlog "github.com/activecm/rita/v5/logger" "github.com/activecm/rita/v5/progressbar" "github.com/activecm/rita/v5/util" "github.com/ClickHouse/clickhouse-go/v2" tea "github.com/charmbracelet/bubbletea" "github.com/google/uuid" ) type HTTPEntry struct { ImportTime time.Time `ch:"import_time"` ZeekUID util.FixedString `ch:"zeek_uid"` Hash util.FixedString `ch:"hash"` Timestamp time.Time `ch:"ts"` Src net.IP `ch:"src"` Dst net.IP `ch:"dst"` SrcNUID uuid.UUID `ch:"src_nuid"` DstNUID uuid.UUID `ch:"dst_nuid"` MultiRequest bool `ch:"multi_request"` SrcPort uint32 `ch:"src_port"` DstPort uint32 `ch:"dst_port"` Duration float64 `ch:"duration"` SrcLocal bool `ch:"src_local"` DstLocal bool `ch:"dst_local"` SrcBytes uint64 `ch:"src_bytes"` DstBytes uint64 `ch:"dst_bytes"` SrcIPBytes uint64 `ch:"src_ip_bytes"` DstIPBytes uint64 `ch:"dst_ip_bytes"` SrcPackets uint64 `ch:"src_packets"` DstPackets uint64 `ch:"dst_packets"` Proto string `ch:"proto"` Service string `ch:"service"` ConnState string `ch:"conn_state"` TransDepth uint64 `ch:"trans_depth"` Method string `ch:"method"` Host string `ch:"host"` URI string `ch:"uri"` Referrer string `ch:"referrer"` HTTPVersion string `ch:"http_version"` UserAgent string `ch:"useragent"` Origin string `ch:"origin"` StatusCode uint64 `ch:"status_code"` StatusMsg string `ch:"status_msg"` InfoCode uint64 `ch:"info_code"` InfoMsg string `ch:"info_msg"` Username string `ch:"username"` Password string `ch:"password"` SrcFUIDs []string `ch:"src_fuids"` SrcFileNames []string `ch:"src_file_names"` SrcMIMETypes []string `ch:"src_mime_types"` DstFUIDs []string `ch:"dst_fuids"` DstFileNames []string `ch:"dst_file_names"` DstMIMETypes []string `ch:"dst_mime_types"` } // parseHTTP listens on a channel of raw http/openhttp log records, formats them and sends them to be linked with conn/openconn records and written to the database func parseHTTP(cfg *config.Config, http <-chan zeektypes.HTTP, output chan database.Data, importTime time.Time, numHTTP *uint64, numConn *uint64) { logger := zlog.GetLogger() // loop over raw http/openhttp channel for h := range http { // parse raw record as an http/open http entry entry, err := formatHTTPRecord(cfg, &h, importTime) if err != nil { logger.Debug().Err(err). Str("log_path", h.LogPath). Str("zeek_uid", h.UID). Str("timestamp", (time.Unix(int64(h.TimeStamp), 0)).String()). Str("src", h.Source). Str("dst", h.Destination). Str("fqdn", h.Host). Str("uri", h.URI). Send() continue } // entry was subject to filtering if entry == nil { continue } if entry.Host == "" { atomic.AddUint64(numConn, 1) } else { atomic.AddUint64(numHTTP, 1) } output <- entry } } // formatHTTPRecord takes a raw http record and formats it into the structure needed by the database func formatHTTPRecord(cfg *config.Config, parseHTTP *zeektypes.HTTP, importTime time.Time) (*HTTPEntry, error) { // get source destination pair for connection record src := parseHTTP.Source dst := parseHTTP.Destination // parse addresses into binary format srcIP := net.ParseIP(src) dstIP := net.ParseIP(dst) // verify that both addresses were able to be parsed successfully if (srcIP == nil) || (dstIP == nil) { return nil, errors.New(errParseSrcDst) } // parse host fqdn := parseHTTP.Host // check if destination is a proxy server based on HTTP method dstIsProxy := (parseHTTP.Method == nethttp.MethodConnect) // if the HTTP method is CONNECT, then the srcIP is communicating // to an FQDN through the dstIP proxy. We need to handle that // as a special case here so that we don't filter internal->internal // connections if the dstIP is an internal IP because the dstIP // is an intermediary and not the final destination. // // The dstIP filter check is not included for proxy connections either // because it isn't really the destination and it doesn't seem to make // sense in this context to check for it. If the proxy IP is external, // this will also allow a user to filter results from other modules // (e.g., beacons), where false positives might arise due to the proxy IP // appearing as a destination, while still allowing for processing that // data for the proxy modules srcLocal := cfg.Filtering.CheckIfInternal(srcIP) dstLocal := cfg.Filtering.CheckIfInternal(dstIP) if dstIsProxy { if cfg.Filtering.FilterDomain(fqdn) || cfg.Filtering.FilterSingleIP(srcIP) { return nil, nil } fqdnAsIPAddress := net.ParseIP(fqdn) if fqdnAsIPAddress != nil && dstLocal && cfg.Filtering.FilterConnPair(srcIP, fqdnAsIPAddress) { return nil, nil } } else if cfg.Filtering.FilterDomain(fqdn) || cfg.Filtering.FilterConnPair(srcIP, dstIP) || // filter out connections where the src is external if the host isn't missing (cfg.Filtering.FilterSNIPair(srcIP) && parseHTTP.Host != "") { return nil, nil } srcNUID := util.ParseNetworkID(srcIP, parseHTTP.AgentUUID) dstNUID := util.ParseNetworkID(dstIP, parseHTTP.AgentUUID) zeekUID, err := util.NewFixedStringHash(parseHTTP.UID) if err != nil { return nil, err } hash, err := util.NewFixedStringHash(srcIP.To16().String(), srcNUID.String(), dstIP.To16().String(), dstNUID.String(), fqdn) if err != nil { return nil, err } entry := &HTTPEntry{ ImportTime: importTime, ZeekUID: zeekUID, Hash: hash, Timestamp: time.Unix(int64(parseHTTP.TimeStamp), 0), Src: srcIP, Dst: dstIP, SrcNUID: srcNUID, DstNUID: dstNUID, SrcPort: parseHTTP.SourcePort, DstPort: parseHTTP.DestinationPort, SrcLocal: srcLocal, DstLocal: dstLocal, TransDepth: parseHTTP.TransDepth, Method: parseHTTP.Method, Host: fqdn, URI: parseHTTP.URI, Referrer: parseHTTP.Referrer, HTTPVersion: parseHTTP.Version, UserAgent: parseHTTP.UserAgent, Origin: parseHTTP.Origin, StatusCode: parseHTTP.StatusCode, StatusMsg: parseHTTP.StatusMsg, InfoCode: parseHTTP.InfoCode, InfoMsg: parseHTTP.InfoMsg, Username: parseHTTP.UserName, Password: parseHTTP.Password, SrcFUIDs: parseHTTP.OrigFuids, SrcFileNames: parseHTTP.OrigFilenames, SrcMIMETypes: parseHTTP.OrigMimeTypes, DstFUIDs: parseHTTP.RespFuids, DstFileNames: parseHTTP.RespFilenames, DstMIMETypes: parseHTTP.RespMimeTypes, } return entry, nil } func (importer *Importer) writeLinkedHTTP(ctx context.Context, progress *tea.Program, barID int, httpWriter, connWriter *database.BulkWriter, open bool) error { logger := zlog.GetLogger() tmpTable := "http_tmp" tableB := "conn_tmp" if open { tmpTable = "openhttp_tmp" tableB = "openconn_tmp" } chCtx := importer.Database.QueryParameters(clickhouse.Parameters{ "tmp_table": tmpTable, "table_b": tableB, }) var totalHTTP uint64 err := importer.Database.Conn.QueryRow(chCtx, ` SELECT count() FROM {tmp_table:Identifier} `).Scan(&totalHTTP) if err != nil { return err } rows, err := importer.Database.Conn.Query(chCtx, ` WITH http_base AS ( SELECT zeek_uid, ts, src, src_nuid, dst, dst_nuid, src_port, dst_port, host, src_local, dst_local, useragent, method, uri, referrer, http_version, trans_depth, origin, status_code, status_msg, info_code, info_msg, username, password, src_fuids, src_file_names, src_mime_types, dst_fuids, dst_mime_types, row_number() OVER (PARTITION BY zeek_uid ORDER BY ts DESC) AS rn FROM {tmp_table:Identifier} ) SELECT h.zeek_uid as zeek_uid, c.hash as hash, c.ts AS ts, h.src_local as src_local, h.dst_local as dst_local, h.src as src, h.src_nuid as src_nuid, h.dst as dst, h.dst_nuid as dst_nuid, h.src_port as src_port, h.dst_port as dst_port, h.host as host, h.useragent as useragent, h.method as method, h.uri as uri, h.referrer as referrer, h.http_version as http_version, h.trans_depth as trans_depth, h.origin as origin, h.status_code as status_code, h.status_msg as status_msg, h.info_code as info_code, h.info_msg as info_msg, h.username as username, h.password as password, h.src_fuids as src_fuids, h.src_file_names as src_file_names, h.src_mime_types as src_mime_types, h.dst_fuids as dst_fuids, h.dst_mime_types as dst_mime_types, -- set proto and service regardless of whether it was linked already or not -- since multi-requests can use different dst ports and still have the same UID, so -- it is useful to be able to see the dst ports coming from multi request entries as well c.proto as proto, c.service as service, if( h.rn = 1, c.src_ip_bytes, 0) as src_ip_bytes, if( h.rn = 1, c.dst_ip_bytes, 0) as dst_ip_bytes, if( h.rn = 1, c.src_bytes, 0) as src_bytes, if( h.rn = 1, c.dst_bytes, 0) as dst_bytes, if( h.rn = 1, c.duration, 0) as duration, if( h.rn = 1, c.conn_state, '') as conn_state, if( h.rn = 1, c.src_packets, 0) as src_packets, if( h.rn = 1, c.dst_packets, 0) as dst_packets, if( h.rn > 1,true, 0) as multi_request FROM http_base h INNER JOIN {table_b:Identifier} c USING zeek_uid WHERE h.rn <= 20 `) if err != nil { return err } i := 0 for rows.Next() { select { // abort this function if the context was cancelled case <-ctx.Done(): logger.Warn().Msg("cancelling HTTP connection linking") rows.Close() return ctx.Err() default: var entry HTTPEntry err := rows.ScanStruct(&entry) if err != nil { return err } i++ // update progress bar every 1000 entries if i%1000 == 0 { progress.Send(progressbar.ProgressMsg{ID: barID, Percent: float64(float64(i) / float64(totalHTTP))}) } entry.ImportTime = importer.Database.ImportStartedAt hash, err := util.NewFixedStringHash(entry.Src.To16().String(), entry.SrcNUID.String(), entry.Host) if err != nil { return err } switch { case entry.Host == "": ignore := importer.Cfg.Filtering.FilterConnPair(entry.Src, entry.Dst) if ignore { continue } icmpType, icmpCode := int64(-1), int64(-1) if entry.Proto == "icmp" { icmpType = int64(entry.SrcPort) icmpCode = int64(entry.DstPort) } httpWriter.WriteChannel <- &entry connEntry := &ConnEntry{ ZeekUID: entry.ZeekUID, ImportID: importer.ImportID, ImportTime: entry.ImportTime, Hash: entry.Hash, Timestamp: entry.Timestamp, Src: entry.Src, Dst: entry.Dst, SrcNUID: entry.SrcNUID, DstNUID: entry.DstNUID, SrcPort: entry.SrcPort, DstPort: entry.DstPort, MissingHostHeader: true, // this field MUST be set to true MissingHostUseragent: entry.UserAgent, // this field MUST be set SrcLocal: entry.SrcLocal, DstLocal: entry.DstLocal, ICMPType: icmpType, ICMPCode: icmpCode, } connWriter.WriteChannel <- connEntry default: entry.Hash = hash httpWriter.WriteChannel <- &entry } } } rows.Close() progress.Send(progressbar.ProgressMsg{ID: barID, Percent: 1}) return nil }