d4-goclient/d4-goclient.go

702 lines
19 KiB
Go

package main
import (
"bytes"
"crypto/hmac"
"crypto/sha256"
"crypto/tls"
"crypto/x509"
"encoding/binary"
"encoding/json"
"flag"
"fmt"
"golang.org/x/net/proxy"
"io"
"io/ioutil"
"log"
"net"
"os"
"os/signal"
"strconv"
"strings"
"syscall"
"time"
config "github.com/D4-project/d4-golang-utils/config"
uuid "github.com/D4-project/d4-golang-utils/crypto/hash"
"github.com/D4-project/d4-golang-utils/inputreader"
"github.com/gomodule/redigo/redis"
)
const (
// VERSION_SIZE
VERSION_SIZE = 1
// TYPE_SIZE
TYPE_SIZE = 1
// UUID_SIZE
UUID_SIZE = 16
// TIMESTAMP_SIZE
TIMESTAMP_SIZE = 8
// HMAC_SIZE
HMAC_SIZE = 32
// SSIZE payload size size
SSIZE = 4
// HDR_SIZE total header size
HDR_SIZE = VERSION_SIZE + TYPE_SIZE + UUID_SIZE + HMAC_SIZE + TIMESTAMP_SIZE + SSIZE
// MH_FILE_LIMIT defines in bytes the max size of the json meta header file
MH_FILE_LIMIT = 100000
)
type (
// A d4 writer implements the io.Writer Interface by implementing Write() and Close()
// it accepts an io.Writer as sink
d4Writer struct {
w io.Writer
key []byte
fb []byte
pb []byte
}
d4S struct {
src io.Reader
dst d4Writer
confdir string
cka time.Duration
ct time.Duration
ce bool
retry time.Duration
rate time.Duration
cc bool
tor bool
daily bool
json bool
ca x509.CertPool
d4error uint8
errnoCopy uint8
debug bool
conf d4params
mhb *bytes.Buffer
mh []byte
redisInputPool *redis.Pool
redisCon redis.Conn
}
d4params struct {
uuid []byte
snaplen uint32
key []byte
version uint8
source string
destination string
ttype uint8
redisHost string
redisPort string
redisQueue string
redisDB int
folderstr string
}
)
var (
// Verbose mode and logging
buf bytes.Buffer
logger = log.New(&buf, "INFO: ", log.Lshortfile)
debugger = log.New(&buf, "DEBUG: ", log.Lmicroseconds)
debugf = func(debug string) {
debugger.Println("", debug)
}
tmpct, _ = time.ParseDuration("5mn")
tmpcka, _ = time.ParseDuration("30s")
tmpretry, _ = time.ParseDuration("30s")
tmprate, _ = time.ParseDuration("200ms")
confdir = flag.String("c", "", "configuration directory")
debug = flag.Bool("v", false, "Set to True, true, TRUE, 1, or t to enable verbose output on stdout - Don't use in production")
ce = flag.Bool("ce", true, "Set to True, true, TRUE, 1, or t to enable TLS on network destination")
ct = flag.Duration("ct", tmpct, "Set timeout in human format")
cka = flag.Duration("cka", tmpcka, "Keep Alive time human format, 0 to disable")
retry = flag.Duration("rt", tmpretry, "Time in human format before retry after connection failure, set to 0 to exit on failure")
rate = flag.Duration("rl", tmprate, "Rate limiter: time in human format before retry after EOF")
cc = flag.Bool("cc", false, "Check TLS certificate against rootCA.crt")
torflag = flag.Bool("tor", false, "Use a SOCKS5 tor proxy on 9050")
dailyflag = flag.Bool("daily", false, "Sets up filewatcher to watch a new %Y%M%D folder at midnight")
jsonflag = flag.Bool("json", false, "The files watched are json files")
)
func main() {
var d4 d4S
d4p := &d4
// Setting up log file
f, err := os.OpenFile("d4-goclient.log", os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
if err != nil {
log.Fatalf("error opening file: %v", err)
}
defer f.Close()
logger.SetOutput(f)
logger.SetFlags(log.LstdFlags | log.Lshortfile)
logger.Println("Init")
flag.Usage = func() {
fmt.Printf("d4 - d4 client\n")
fmt.Printf("Read data from the configured <source> and send it to <destination>\n")
fmt.Printf("\n")
fmt.Printf("Usage: d4 -c config_directory\n")
fmt.Printf("\n")
fmt.Printf("Configuration\n\n")
fmt.Printf("The configuration settings are stored in files in the configuration directory\n")
fmt.Printf("specified with the -c command line switch.\n\n")
fmt.Printf("Files in the configuration directory\n")
fmt.Printf("\n")
fmt.Printf("key - is the private HMAC-SHA-256-128 key.\n")
fmt.Printf(" The HMAC is computed on the header with a HMAC value set to 0\n")
fmt.Printf(" which is updated later.\n")
fmt.Printf("snaplen - the length of bytes that is read from the <source>\n")
fmt.Printf("version - the version of the d4 client\n")
fmt.Printf("type - the type of data that is send. pcap, netflow, ...\n")
fmt.Printf("source - the source where the data is read from\n")
fmt.Printf("destination - the destination where the data is written to\n")
fmt.Printf("redis_d4 - location of redis d4 server\n")
fmt.Printf("redis_queue - analyzer:type:queueuuid to pop\n")
fmt.Printf("\n")
flag.PrintDefaults()
}
flag.Parse()
if flag.NFlag() == 0 || *confdir == "" {
flag.Usage()
os.Exit(1)
} else {
*confdir = strings.TrimSuffix(*confdir, "/")
*confdir = strings.TrimSuffix(*confdir, "\\")
}
d4.confdir = *confdir
d4.ce = *ce
d4.ct = *ct
d4.cc = *cc
d4.json = *jsonflag
d4.cka = *cka
d4.retry = *retry
d4.rate = *rate
d4.tor = *torflag
d4.daily = *dailyflag
s := make(chan os.Signal, 1)
signal.Notify(s, os.Interrupt, os.Kill)
errchan := make(chan error)
eofchan := make(chan string)
metachan := make(chan string)
// Launching the Rate limiters
rateLimiter := time.Tick(d4.rate)
retryLimiter := time.Tick(d4.retry)
//Setup
if !d4loadConfig(d4p) {
panic("Could not load Config.")
}
if !setReaderWriters(d4p, false) {
panic("Could not Init Inputs Outputs.")
}
if !d4.dst.initHeader(d4p) {
panic("Could not Init Headers.")
}
// force is a flag that forces the creation of a new connection
force := false
// On the first run, we send d4 meta header for type 2/254
if d4.conf.ttype == 254 || d4.conf.ttype == 2 {
go sendMeta(d4p, errchan, metachan)
H:
for {
select {
case <-errchan:
select {
case <-retryLimiter:
go sendMeta(d4p, errchan, metachan)
case <-s:
logger.Println("Exiting")
exit(d4p, 0)
}
case <-metachan:
break H
}
}
}
// Launch copy routine
go d4Copy(d4p, errchan, eofchan)
// Handle signals
for {
select {
// Case where the input ran out of data to consume1
case <-eofchan:
// We wait for ratelimiter before polling again
EOF:
for {
select {
case <-rateLimiter:
// copy routine
go d4Copy(d4p, errchan, eofchan)
break EOF
// Exit signal
case <-s:
logger.Println("Exiting")
exit(d4p, 0)
}
}
// ERROR, we check first whether it is network related
case err := <-errchan:
// On connection errors, we force setReaderWriter to reset the connection
force = false
switch t := err.(type) {
case *net.OpError:
force = true
if t.Op == "dial" {
logger.Println("Unknown Host")
} else if t.Op == "read" {
logger.Println("Connection Refused")
} else if t.Op == "write" {
logger.Println("Write error")
}
case syscall.Errno:
if t == syscall.ECONNREFUSED {
force = true
logger.Println("Connection Refused")
}
}
// We wait for retryLimiter before writing again
RETRY:
for {
select {
case <-retryLimiter:
if !setReaderWriters(d4p, force) {
// Can't connect, we break to retry
// force is still true
break
}
if !d4.dst.initHeader(d4p) {
panic("Could not Init Headers.")
}
if (d4.conf.ttype == 254 || d4.conf.ttype == 2) && force {
// setReaderWriter is happy, we should have a working
// connection from now on.
force = false
// Sending meta header for the first time on this new connection
go sendMeta(d4p, errchan, metachan)
}
break RETRY
// Exit signal
case <-s:
logger.Println("Exiting")
exit(d4p, 0)
}
}
// metaheader sent, launch the copy routine
case <-metachan:
go d4Copy(d4p, errchan, eofchan)
// Exit signal
case <-s:
logger.Println("Exiting")
exit(d4p, 0)
}
}
}
func exit(d4 *d4S, exitcode int) {
// Output debug info in the log before closing if debug is enabled
if *debug == true {
(*d4).debug = true
fmt.Print(&buf)
}
os.Exit(exitcode)
}
func d4Copy(d4 *d4S, errchan chan error, eofchan chan string) {
nread, err := io.CopyBuffer(&d4.dst, d4.src, d4.dst.pb)
// Always retry
if err != nil {
logger.Printf("D4copy: %s", err)
errchan <- err
return
}
eofchan <- fmt.Sprintf("EOF: Nread: %d", nread)
}
func sendMeta(d4 *d4S, errchan chan error, metachan chan string) {
// Fill metaheader buffer with metaheader data
d4.mhb = bytes.NewBuffer(d4.mh)
d4.dst.hijackHeader()
// Ugly hack to skip bytes.Buffer WriteTo check that bypasses my fixed lenght buffer
nread, err := io.CopyBuffer(&d4.dst, struct{ io.Reader }{d4.mhb}, d4.dst.pb)
if err != nil {
logger.Printf("Cannot sent meta-header: %s", err)
errchan <- err
return
}
logger.Println(fmt.Sprintf("Meta-Header sent: %d bytes", nread))
d4.dst.restoreHeader()
metachan <- "Header Sent"
return
}
func readConfFile(d4 *d4S, fileName string) []byte {
return config.ReadConfigFile((*d4).confdir, fileName)
}
func d4loadConfig(d4 *d4S) bool {
// populate the map
(*d4).conf = d4params{}
(*d4).conf.source = string(readConfFile(d4, "source"))
if len((*d4).conf.source) < 1 {
log.Fatal("Unsupported source")
}
if (*d4).conf.source == "folder" {
fstr := string(readConfFile(d4, "folder"))
if ffd, err := os.Stat(fstr); os.IsNotExist(err) {
log.Fatal("Folder does not exist")
} else {
if !ffd.IsDir() {
log.Fatal("Folder is not a directory")
}
}
(*d4).conf.folderstr = fstr
}
if (*d4).conf.source == "d4server" {
// Parse Input Redis Config
tmp := string(readConfFile(d4, "redis_d4"))
ss := strings.Split(string(tmp), "/")
if len(ss) <= 1 {
log.Fatal("Missing Database in Redis input config: should be host:port/database_name")
}
(*d4).conf.redisDB, _ = strconv.Atoi(ss[1])
var ret bool
ret, ss[0] = config.IsNet(ss[0])
if ret {
sss := strings.Split(string(ss[0]), ":")
(*d4).conf.redisHost = sss[0]
(*d4).conf.redisPort = sss[1]
} else {
log.Fatal("Redis config error.")
}
(*d4).conf.redisQueue = string(config.ReadConfigFile(*confdir, "redis_queue"))
}
(*d4).conf.destination = string(readConfFile(d4, "destination"))
if len((*d4).conf.destination) < 1 {
log.Fatal("Unsupported Destination")
}
tmpu, err := uuid.FromString(string(readConfFile(d4, "uuid")))
if err != nil {
// generate new uuid
(*d4).conf.uuid = generateUUIDv4()
// And push it into the conf file
f, err := os.OpenFile((*d4).confdir+"/uuid", os.O_WRONLY, 0666)
defer f.Close()
if err != nil {
log.Fatal(err)
}
// store as canonical representation
f.WriteString(fmt.Sprintf("%s", uuid.FromBytesOrNil((*d4).conf.uuid)) + "\n")
} else {
(*d4).conf.uuid = tmpu.Bytes()
}
// parse snaplen to uint32
tmp, err := strconv.ParseUint(string(readConfFile(d4, "snaplen")), 10, 32)
if err != nil || tmp < 1 {
(*d4).conf.snaplen = uint32(4096)
} else {
(*d4).conf.snaplen = uint32(tmp)
}
(*d4).conf.key = readConfFile(d4, "key")
// parse version to uint8
tmp, _ = strconv.ParseUint(string(readConfFile(d4, "version")), 10, 8)
if err != nil || tmp < 1 {
(*d4).conf.version = uint8(1)
} else {
(*d4).conf.version = uint8(tmp)
}
// parse type to uint8
tmp, _ = strconv.ParseUint(string(readConfFile(d4, "type")), 10, 8)
if err != nil || tmp < 1 {
log.Fatal("Unsupported type")
} else {
(*d4).conf.ttype = uint8(tmp)
}
// parse meta header file
data := make([]byte, MH_FILE_LIMIT)
if tmp == 254 || tmp == 2 {
file, err := os.Open((*d4).confdir + "/metaheader.json")
defer file.Close()
if err != nil {
panic("Failed to open Meta-Header File.")
} else {
if count, err := file.Read(data); err != nil {
panic("Failed to open Meta-Header File.")
} else {
if json.Valid(data[:count]) {
if checkType(data[:count]) {
if off, err := file.Seek(0, 0); err != nil || off != 0 {
panic(fmt.Sprintf("Cannot read Meta-Header file: %s", err))
} else {
// create metaheader buffer
d4.mhb = bytes.NewBuffer(d4.mh)
if err := json.Compact((*d4).mhb, data[:count]); err != nil {
logger.Println("Failed to compact meta header file")
}
// Store the metaheader in d4 struct for subsequent retries
(*d4).mh = data[:count]
}
} else {
panic("A Meta-Header File should at least contain a 'type' field.")
}
} else {
panic("Failed to validate open Meta-Header File.")
}
}
}
}
// Add the custom CA cert in D4 certpool
if (*d4).cc {
certb, _ := ioutil.ReadFile((*d4).confdir + "rootCA.crt")
(*d4).ca = *x509.NewCertPool()
ok := (*d4).ca.AppendCertsFromPEM(certb)
if !ok {
panic("Failed to parse provided root certificate.")
}
}
return true
}
func checkType(b []byte) bool {
var f interface{}
if err := json.Unmarshal(b, &f); err != nil {
return false
}
m := f.(map[string]interface{})
for k, v := range m {
if k == "type" {
switch v.(type) {
case string:
if v != nil {
return true
}
}
}
}
return false
}
func newD4Writer(writer io.Writer, key []byte) d4Writer {
return d4Writer{w: writer, key: key}
}
// TODO QUICK IMPLEM, REVISE
func setReaderWriters(d4 *d4S, force bool) bool {
//TODO implement other destination file, fifo unix_socket ...
switch (*d4).conf.source {
case "stdin":
(*d4).src = os.Stdin
case "pcap":
f, _ := os.Open("capture.pcap")
(*d4).src = f
case "d4server":
// Create a new redis connection pool
(*d4).redisInputPool = newPool((*d4).conf.redisHost+":"+(*d4).conf.redisPort, 16)
var err error
(*d4).redisCon, err = (*d4).redisInputPool.Dial()
if err != nil {
logger.Println("Could not connect to d4 Redis")
return false
}
(*d4).src, err = inputreader.NewLPOPReader(&(*d4).redisCon, (*d4).conf.redisDB, (*d4).conf.redisQueue)
if err != nil {
log.Printf("Could not create d4 Redis Descriptor %q \n", err)
return false
}
case "folder":
var err error
(*d4).src, err = inputreader.NewFileWatcherReader((*d4).conf.folderstr, (*d4).json, (*d4).daily, logger)
if err != nil {
log.Printf("Could not create File Watcher %q \n", err)
return false
}
}
isn, dstnet := config.IsNet((*d4).conf.destination)
if isn {
// We test whether a connection already exist
// (case where the reader run out of data)
// force forces to reset the connections after
// failure to reuse it
if _, ok := (*d4).dst.w.(net.Conn); !ok || force {
if (*d4).tor {
dialer := net.Dialer{
Timeout: (*d4).ct,
KeepAlive: (*d4).cka,
FallbackDelay: 0,
}
dial, err := proxy.SOCKS5("tcp", "127.0.0.1:9050", nil, &dialer)
if err != nil {
log.Fatal(err)
}
tlsc := tls.Config{
InsecureSkipVerify: true,
}
if (*d4).cc {
tlsc = tls.Config{
InsecureSkipVerify: false,
RootCAs: &(*d4).ca,
}
}
conn, errc := dial.Dial("tcp", dstnet)
if errc != nil {
logger.Println(errc)
return false
}
if (*d4).ce == true {
conn = tls.Client(conn, &tlsc) // use tls
}
(*d4).dst = newD4Writer(conn, (*d4).conf.key)
} else {
dial := net.Dialer{
Timeout: (*d4).ct,
KeepAlive: (*d4).cka,
FallbackDelay: 0,
}
tlsc := tls.Config{
InsecureSkipVerify: true,
}
if (*d4).cc {
tlsc = tls.Config{
InsecureSkipVerify: false,
RootCAs: &(*d4).ca,
}
}
if (*d4).ce == true {
conn, errc := tls.DialWithDialer(&dial, "tcp", dstnet, &tlsc)
if errc != nil {
logger.Println(errc)
return false
}
(*d4).dst = newD4Writer(conn, (*d4).conf.key)
} else {
conn, errc := dial.Dial("tcp", dstnet)
if errc != nil {
return false
}
(*d4).dst = newD4Writer(conn, (*d4).conf.key)
}
}
}
} else {
switch (*d4).conf.destination {
case "stdout":
(*d4).dst = newD4Writer(os.Stdout, (*d4).conf.key)
case "file":
f, _ := os.Create("test.txt")
(*d4).dst = newD4Writer(f, (*d4).conf.key)
default:
panic(fmt.Sprintf("No suitable destination found, given :%q", (*d4).conf.destination))
}
}
// Create the copy buffer
(*d4).dst.fb = make([]byte, HDR_SIZE+(*d4).conf.snaplen)
(*d4).dst.pb = make([]byte, (*d4).conf.snaplen)
return true
}
func generateUUIDv4() []byte {
uuid, err := uuid.NewV4()
if err != nil {
log.Fatal(err)
}
logger.Println(fmt.Sprintf("UUIDv4: %s", uuid))
return uuid.Bytes()
}
func (d4w *d4Writer) Write(bs []byte) (int, error) {
// bs is pb
// zero out moving parts of the frame
copy(d4w.fb[18:62], make([]byte, 44))
copy(d4w.fb[62:], make([]byte, 62+len(bs)))
// update headers
d4w.updateHeader(len(bs))
// Copy payload after the header
copy(d4w.fb[62:62+len(bs)], bs)
// Now that the packet is complete, compute hmac
d4w.updateHMAC(len(bs))
// Eventually write binary in the sink
err := binary.Write(d4w.w, binary.LittleEndian, d4w.fb[:62+len(bs)])
return len(bs), err
}
// TODO write go idiomatic err return values
func (d4w *d4Writer) updateHeader(lenbs int) bool {
timeUnix := time.Now().Unix()
binary.LittleEndian.PutUint64(d4w.fb[18:26], uint64(timeUnix))
binary.LittleEndian.PutUint32(d4w.fb[58:62], uint32(lenbs))
return true
}
func (d4w *d4Writer) updateHMAC(ps int) bool {
h := hmac.New(sha256.New, d4w.key)
h.Write(d4w.fb[0:1])
h.Write(d4w.fb[1:2])
h.Write(d4w.fb[2:18])
h.Write(d4w.fb[18:26])
h.Write(make([]byte, 32))
h.Write(d4w.fb[58:62])
h.Write(d4w.fb[62 : 62+ps])
copy(d4w.fb[26:58], h.Sum(nil))
return true
}
func (d4w *d4Writer) initHeader(d4 *d4S) bool {
// zero out the header
copy(d4w.fb[:HDR_SIZE], make([]byte, HDR_SIZE))
// put version and type into the header
d4w.fb[0] = (*d4).conf.version
d4w.fb[1] = (*d4).conf.ttype
// put uuid into the header
copy(d4w.fb[2:18], (*d4).conf.uuid)
// timestamp
timeUnix := time.Now().UnixNano()
binary.LittleEndian.PutUint64(d4w.fb[18:26], uint64(timeUnix))
// hmac is set to zero during hmac operations, so leave it alone
// init size of payload at 0
binary.LittleEndian.PutUint32(d4w.fb[58:62], uint32(0))
debugf(fmt.Sprintf("Initialized a %d bytes header:\n", HDR_SIZE))
debugf(fmt.Sprintf("%b\n", d4w.fb[:HDR_SIZE]))
return true
}
// We use type 2 to send the meta header
func (d4w *d4Writer) hijackHeader() bool {
d4w.fb[1] = 2
return true
}
// Switch back the header to 254
func (d4w *d4Writer) restoreHeader() bool {
d4w.fb[1] = 254
return true
}
func newPool(addr string, maxconn int) *redis.Pool {
return &redis.Pool{
MaxActive: maxconn,
MaxIdle: 3,
IdleTimeout: 240 * time.Second,
// Dial or DialContext must be set. When both are set, DialContext takes precedence over Dial.
Dial: func() (redis.Conn, error) { return redis.Dial("tcp", addr) },
}
}