activecm-rita/database/db.go

package database

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log"
	"net"
	"time"

	"github.com/activecm/rita/v5/config"
	zlog "github.com/activecm/rita/v5/logger"

	clickhouse "github.com/ClickHouse/clickhouse-go/v2"
	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
)

var ErrInvalidDatabaseConnection = fmt.Errorf("database connection is nil")
var ErrInvalidMinMaxTimestamp = fmt.Errorf("invalid min or max timestamp")

// DB is the workhorse container for messing with the database
type DB struct {
	Conn            driver.Conn
	selected        string
	Rolling         bool
	rebuild         bool
	ctx             context.Context
	cancel          context.CancelFunc
	ImportStartedAt time.Time
}

// GetSelectedDB returns the name of the target database of db connection
func (db *DB) GetSelectedDB() string {
	return db.selected
}

// QueryParameters generates ClickHouse query parameters by creating a context with the specified parameters in it
func (db *DB) QueryParameters(params clickhouse.Parameters) context.Context {
	return clickhouse.Context(db.ctx, clickhouse.WithParameters(params))
}

// GetContext returns the context for the database connection
func (db *DB) GetContext() context.Context {
	return db.ctx
}

// getConn returns the driver connection
func (db *DB) getConn() driver.Conn {
	return db.Conn
}

func (db *DB) GetBeaconMinMaxTimestamps() (time.Time, time.Time, bool, error) {

	var minTS, maxTS time.Time
	var notFromConn bool

	if db.Conn == nil {
		return time.Unix(0, 0), time.Unix(0, 0), notFromConn, ErrInvalidDatabaseConnection
	}

	logger := zlog.GetLogger()

	ctx := db.QueryParameters(clickhouse.Parameters{
		"database": db.selected,
	})
	// min timestamp: max timestamp - 24 hours, capped to the actual minimum timestamp from the logs
	// max timestamp: max timestamp in the logs
	err := db.Conn.QueryRow(ctx, `
		SELECT greatest(min_ts, timestamp_sub(HOUR, 24, max_ts)) as min_ts, max_ts FROM (
			SELECT min(min_ts) AS min_ts, max(max_ts) AS max_ts FROM metadatabase.min_max
			WHERE database = {database:String} AND beacon = true
			GROUP BY database
		)
	`).Scan(&minTS, &maxTS)

	// return error if the error is not a no rows found error
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		logger.Err(err).Str("database", db.selected).Msg("failed to get max ts from metadatabase min_max table")
		return time.Unix(0, 0), time.Unix(0, 0), notFromConn, err
	}

	if maxTS.IsZero() {
		return time.Unix(0, 0), time.Unix(0, 0), notFromConn, ErrInvalidMinMaxTimestamp
	}
	if minTS.IsZero() {
		return time.Unix(0, 0), time.Unix(0, 0), notFromConn, ErrInvalidMinMaxTimestamp
	}

	// if dataset is not rolling or if the max timestamp is over 24 hours ago, use the max timestamp
	return minTS, maxTS, notFromConn, nil

}

func (db *DB) GetTrueMinMaxTimestamps() (time.Time, time.Time, bool, bool, error) {
	logger := zlog.GetLogger()

	var minTS, maxTS time.Time
	var notFromConn bool
	var useCurrentTime bool

	if db.Conn == nil {
		return time.Unix(0, 0), time.Unix(0, 0), false, false, ErrInvalidDatabaseConnection
	}

	rolling, err := GetRollingStatus(db.GetContext(), db.Conn, db.GetSelectedDB())
	if err != nil && !errors.Is(err, ErrDatabaseNotFound) {
		return time.Unix(0, 0), time.Unix(0, 0), false, false, err
	}
	if errors.Is(err, ErrDatabaseNotFound) {
		rolling = db.Rolling
	}

	ctx := db.QueryParameters(clickhouse.Parameters{
		"database": db.selected,
	})
	// min timestamp: max timestamp - 24 hours, capped to the actual minimum timestamp from the logs
	// max timestamp: max timestamp in the logs
	err = db.Conn.QueryRow(ctx, `
		SELECT greatest(min_ts, timestamp_sub(HOUR, 24, max_ts)) as min_ts, max_ts FROM (
			SELECT min(min_ts) AS min_ts, max(max_ts) AS max_ts FROM metadatabase.min_max
			WHERE database = {database:String} 
			GROUP BY database
		)
	`).Scan(&minTS, &maxTS)

	// return error if the error is not a no rows found error
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		logger.Err(err).Str("database", db.selected).Msg("failed to get max ts from metadatabase min_max table")
		return time.Unix(0, 0), time.Unix(0, 0), notFromConn, useCurrentTime, err
	}

	if maxTS.IsZero() {
		return time.Unix(0, 0), time.Unix(0, 0), notFromConn, useCurrentTime, fmt.Errorf("could not find any viable max timestamp")
	}
	if minTS.IsZero() {
		return time.Unix(0, 0), time.Unix(0, 0), notFromConn, useCurrentTime, fmt.Errorf("could not find any viable min timestamp")
	}

	// if dataset is rolling and the max timestamp is not over 24 hours ago, use the current time for first seen
	if rolling && time.Since(maxTS).Hours() <= 24 {
		useCurrentTime = true
	}

	// if dataset is not rolling or if the max timestamp is over 24 hours ago, use the max timestamp
	return minTS, maxTS, notFromConn, useCurrentTime, nil

}

// GetNetworkSize returns the number of distinct internal hosts for the past 24 hours, which is used to determine prevalence
func (db *DB) GetNetworkSize(minTS time.Time) (uint64, error) {
	logger := zlog.GetLogger()

	var networkSize uint64

	ctx := db.QueryParameters(clickhouse.Parameters{
		"min_ts": fmt.Sprintf("%d", minTS.UTC().Unix()),
	})

	err := db.Conn.QueryRow(ctx, `
		SELECT count() FROM (
			-- uconn
			SELECT DISTINCT src FROM uconn
			WHERE src_local = true AND hour >= toStartOfHour(fromUnixTimestamp({min_ts:Int64}))
			UNION DISTINCT
			SELECT DISTINCT dst AS src FROM uconn
			WHERE dst_local = true AND hour >= toStartOfHour(fromUnixTimestamp({min_ts:Int64}))
			UNION DISTINCT
			-- openconn
			SELECT DISTINCT src FROM openconn
			WHERE src_local = true
			UNION DISTINCT
			SELECT DISTINCT dst AS src FROM openconn
			WHERE dst_local = true
			UNION DISTINCT
			-- http
			SELECT DISTINCT src FROM usni
			WHERE http = true AND src_local = true AND hour >= toStartOfHour(fromUnixTimestamp({min_ts:Int64}))
			UNION DISTINCT
			SELECT DISTINCT dst AS src FROM usni
			WHERE http = true AND dst_local = true AND hour >= toStartOfHour(fromUnixTimestamp({min_ts:Int64}))
			UNION DISTINCT
			-- openhttp
			SELECT DISTINCT src FROM openhttp
			WHERE src_local = true
			UNION DISTINCT
			SELECT DISTINCT dst AS src FROM openhttp
			WHERE dst_local = true
			UNION DISTINCT
			-- dns
			SELECT DISTINCT src FROM udns
			WHERE src_local = true AND hour >= toStartOfHour(fromUnixTimestamp({min_ts:Int64}))
			UNION DISTINCT
			SELECT DISTINCT dst AS src FROM udns
			WHERE dst_local = true AND hour >= toStartOfHour(fromUnixTimestamp({min_ts:Int64}))
		)
	`).Scan(&networkSize)

	if err != nil {
		logger.Err(err).Str("database", db.selected).Msg("failed to network size from uconn table")
		return networkSize, err
	}

	return networkSize, nil
}

// TruncateTmpLinkTables truncates the tables that are used to link zeek uids.
// This should be called after each import so that these tmp tables don't take up unnecessary disk space.
func (db *DB) TruncateTmpLinkTables() error {
	ctx := db.QueryParameters(clickhouse.Parameters{
		"database": db.selected,
	})

	if err := db.Conn.Exec(ctx, `--sql
		TRUNCATE TABLE IF EXISTS {database:Identifier}.conn_tmp
	`); err != nil {
		return err
	}

	if err := db.Conn.Exec(ctx, `--sql
		TRUNCATE TABLE IF EXISTS {database:Identifier}.ssl_tmp
	`); err != nil {
		return err
	}

	if err := db.Conn.Exec(ctx, `--sql
		TRUNCATE TABLE IF EXISTS {database:Identifier}.http_tmp
	`); err != nil {
		return err
	}

	if err := db.Conn.Exec(ctx, `--sql
		TRUNCATE TABLE IF EXISTS {database:Identifier}.openconn_tmp
	`); err != nil {
		return err
	}

	if err := db.Conn.Exec(ctx, `--sql
		TRUNCATE TABLE IF EXISTS {database:Identifier}.openssl_tmp
	`); err != nil {
		return err
	}

	if err := db.Conn.Exec(ctx, `--sql
		TRUNCATE TABLE IF EXISTS {database:Identifier}.openhttp_tmp
	`); err != nil {
		return err
	}
	return nil
}

// ResetTemporaryTables clears out data in tmp tables (if they exist) from the previous import
func (db *DB) ResetTemporaryTables() error {
	ctx := db.QueryParameters(clickhouse.Parameters{
		"database": db.selected,
	})

	if err := db.Conn.Exec(ctx, `--sql
		TRUNCATE TABLE IF EXISTS {database:Identifier}.openconn
	`); err != nil {
		return err
	}

	if err := db.Conn.Exec(ctx, `--sql
		TRUNCATE TABLE IF EXISTS {database:Identifier}.openhttp
	`); err != nil {
		return err
	}

	if err := db.Conn.Exec(ctx, `--sql
		TRUNCATE TABLE IF EXISTS {database:Identifier}.openssl
	`); err != nil {
		return err
	}

	if err := db.Conn.Exec(ctx, `--sql
		TRUNCATE TABLE IF EXISTS {database:Identifier}.uconn_tmp
	`); err != nil {
		return err
	}

	if err := db.Conn.Exec(ctx, `--sql
		TRUNCATE TABLE IF EXISTS {database:Identifier}.openconnhash_tmp
	`); err != nil {
		return err
	}

	if err := db.Conn.Exec(ctx, `--sql
		TRUNCATE TABLE IF EXISTS {database:Identifier}.opensniconn_tmp
	`); err != nil {
		return err
	}

	if err := db.Conn.Exec(ctx, `--sql
		TRUNCATE TABLE IF EXISTS {database:Identifier}.sniconn_tmp
	`); err != nil {
		return err
	}

	if err := db.Conn.Exec(ctx, `--sql
		TRUNCATE TABLE IF EXISTS {database:Identifier}.dns_tmp
	`); err != nil {
		return err
	}

	return db.TruncateTmpLinkTables()

}

// ConnectToDB sets up a new connection to the specified database
func ConnectToDB(ctx context.Context, db string, cfg *config.Config, cancel context.CancelFunc) (*DB, error) {
	// connect to the database
	conn, err := clickhouse.Open(&clickhouse.Options{
		Addr: []string{cfg.Env.DBConnection},
		Auth: clickhouse.Auth{
			Database: db,
			Username: cfg.Env.DBUsername,
			Password: cfg.Env.DBPassword,
		},
		DialContext: func(ctx context.Context, addr string) (net.Conn, error) {
			// dialCount++
			var d net.Dialer
			return d.DialContext(ctx, "tcp", addr)
		},
		Debug: false,
		Debugf: func(format string, v ...any) {
			log.Println(format, v)
		},
		Settings: clickhouse.Settings{
			"max_execution_time": cfg.RITA.MaxQueryExecutionTime,
			"mutations_sync":     1,
		},
		Compression: &clickhouse.Compression{
			Method: clickhouse.CompressionLZ4,
		},
		DialTimeout:          time.Second * 120,
		MaxOpenConns:         50,
		MaxIdleConns:         50,
		ConnMaxLifetime:      time.Duration(1) * time.Hour,
		ConnOpenStrategy:     clickhouse.ConnOpenInOrder,
		BlockBufferSize:      10,
		MaxCompressionBuffer: 10240,

		ClientInfo: clickhouse.ClientInfo{ // optional, please see Client info section in the clickhouse-go README.md
			Products: []struct {
				Name    string
				Version string
			}{
				{Name: "rita", Version: "0.1"},
			},
		},
	})

	// check if the connection call had any errors
	if err != nil {
		return nil, err
	}

	// check if the connection is valid
	if err := conn.Ping(ctx); err != nil {
		// if exception, ok := err.(*clickhouse.Exception); ok {
		// 	fmt.Printf("Exception [%d] %s \n%s\n", exception.Code, exception.Message, exception.StackTrace)
		// }
		return nil, err
	}

	// fmt.Println("Validated connection to database", db)

	return &DB{
		Conn:     conn,
		ctx:      ctx,
		cancel:   cancel,
		selected: db,
	}, nil
}

// GetFirstSeenTimestamp gets the relative timestamp to use for calculating/displaying first seen.
// Returns max timestamp, whether or not to use the current time, and error
// func (db *DB) GetFirstSeenTimestamp() (time.Time, time.Time, bool, error) {
// 	rolling, err := GetRollingStatus(db.GetContext(), db.Conn, db.GetSelectedDB())
// 	if err != nil {
// 		return time.Unix(0, 0), time.Unix(0, 0), false, err
// 	}

// 	minTS, maxTS, _, err := db.GetMinMaxTimestamps()
// 	if err != nil {
// 		return time.Unix(0, 0), time.Unix(0, 0), false, fmt.Errorf("could not get min/max timestamps for analysis: %w", err)
// 	}

// 	// if dataset is not rolling or if the max timestamp is over 24 hours ago, use the max timestamp
// 	if !rolling || time.Since(maxTS).Hours() > 24 {
// 		return maxTS, minTS, false, nil
// 	}

// 	// if rolling and maxTS <= 24 hrs ago, use the current time
// 	return time.Unix(0, 0), time.Unix(0, 0), true, nil
// }