logicmonclient/logicmonclient.go

/*
This Source Code Form is subject to the terms of the Mozilla Public
License, v. 2.0. If a copy of the MPL was not distributed with this
file, You can obtain one at https://mozilla.org/MPL/2.0/.
*/

package logicmonclient

// https://www.logicmonitor.com/support/rest-api-developers-guide/overview/using-logicmonitors-rest-api
// https://www.logicmonitor.com/swagger-ui-master/api-v3/lm-sdkv3-docs.html

import (
	"context"
	"crypto/hmac"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/url"
	"strconv"
	"time"

	dp "idio.link/go/depager"
)

var Debug = false

// LogicMonitor REST API v3 supports maximum size of 1000
// records. Even if you specify a number above 1000, the
// response will always return 1000 records.
const MaxPageSize = 1000

func debug(format string, msg ...any) {
	if Debug {
		log.Printf("DEBUG: "+format, msg...)
	}
}

type LMv1TokenBasedCredential struct {
	AccessId  string
	AccessKey string
}

func NewLMv1TokenBasedCredential(
	accessId,
	accessKey string,
) *LMv1TokenBasedCredential {
	return &LMv1TokenBasedCredential{accessId, accessKey}
}

func generateAuthorization(
	timestamp,
	httpMethod,
	resource string,
	requestBody []byte,
	cred *LMv1TokenBasedCredential,
) string {
	/* From https://www.logicmonitor.com/support/rest-api-authentication:

	   signature =
	     base64(HMAC-SHA256(
	       Access Key,
	       HTTP VERB +
	         TIMESTAMP (in epoch milliseconds) +
	         POST/PUT DATA (if any) +
	      	 RESOURCE PATH,
	     ))

	   Authorization: LMv1 AccessId:Signature:Timestamp
	*/
	mac := hmac.New(sha256.New, []byte(cred.AccessKey))
	mac.Write([]byte(httpMethod))
	mac.Write([]byte(timestamp))
	mac.Write(requestBody)
	mac.Write([]byte(resource))
	sum := hex.EncodeToString(mac.Sum(nil))
	sig := base64.StdEncoding.EncodeToString([]byte(sum))
	auth := fmt.Sprintf(
		"LMv1 %s:%s:%s",
		cred.AccessId,
		sig,
		timestamp,
	)
	debug("authorization: %v", auth)
	return auth
}

func NewLMClient(
	accountName string,
	ctx context.Context,
	httpClient *http.Client,
	credential *LMv1TokenBasedCredential,
) *LMClient {
	return &LMClient{
		accountName: accountName,
		domain:      "logicmonitor.com",
		basePath:    "/santaba/rest",
		ctx:         ctx,
		httpClient:  httpClient,
		credential:  credential,
		retries:     3,
		retryWait:   10 * time.Second,
	}
}

type LMClient struct {
	accountName     string
	domain          string
	basePath        string
	ctx             context.Context
	httpClient      *http.Client
	credential      *LMv1TokenBasedCredential
	blockReqsBefore time.Time
	retries         int
	retryWait       time.Duration
}

func (c *LMClient) request(
	method string,
	uri *url.URL,
	body io.Reader,
) (respBody []byte, err error) {
	uri.Scheme = "https"
	uri.Host = c.accountName + "." + c.domain

	resource := uri.EscapedPath() // save for authentication

	path, err := url.JoinPath(c.basePath, uri.EscapedPath())
	if err != nil {
		panic(fmt.Sprintf("BUG: lmclient: request: failed to join path: %s + %s", c.basePath, uri.RequestURI()))
	}
	uri.Path = path
	debug("request %s", uri)

	req, err :=
		http.NewRequestWithContext(
			c.ctx,
			method,
			uri.String(),
			body,
		)
	if err != nil {
		err = fmt.Errorf("request %v: %v", uri, err)
		return
	}

	var reqBody []byte
	if req.Body != nil {
		reqBody, err = io.ReadAll(req.Body)
		if err != nil {
			err = fmt.Errorf("request %v: failed to read request body: %v", uri, err)
			return
		}
	}

	// Only support API version 3
	req.Header.Set("X-version", "3")
	req.Header.Set("User-Agent", "logicmonclient")

	/* From https://www.logicmonitor.com/support/rest-api-authentication:

	> When LogicMonitor servers receive an API request, they
	> ensure the specified timestamp is within 30 minutes of
	> the current time.

	In the worst case, the below logic will fail long before
	the timestamp disparity reaches 30 minutes.
	*/
	timestamp := strconv.FormatInt(time.Now().UnixMilli(), 10)
	auth := generateAuthorization(
		timestamp,
		req.Method,
		resource,
		reqBody,
		c.credential,
	)
	req.Header.Set("Authorization", auth)

	// retry when server cannot handle request
	var resp *http.Response
	for i := 0; i < c.retries+1; i++ {
		resp, err = c.httpClient.Do(req)
		if err != nil {
			err = fmt.Errorf("request %v: %v", uri, err)
			return
		}
		if resp.StatusCode != http.StatusTooManyRequests {
			break
		}
		// Offer an async option in future, but this is
		// acceptable, for now.
		time.Sleep(c.retryWait / time.Millisecond)
	}
	defer resp.Body.Close()

	respBody, err = io.ReadAll(resp.Body)
	if err != nil {
		err = fmt.Errorf("request %v: failed to read response body: %v", uri, err)
		return
	}

	// Throttle requests
	// See https://www.logicmonitor.com/support/rest-api-developers-guide/overview/rate-limiting
	remStr := resp.Header.Get("X-Rate-Limit-Remaining")
	winStr := resp.Header.Get("X-Rate-Limit-Window")
	remaining, err := strconv.ParseInt(remStr, 10, 64)
	if remStr != "" && err != nil {
		err = fmt.Errorf("request %v: failed to parse header X-Rate-Limit-Remaining: %v", uri, err)
		return
	}
	window, err := time.ParseDuration(winStr + "s")
	if winStr != "" && err != nil {
		err = fmt.Errorf("request %v: failed to parse header X-Rate-Limit-Window: %v", uri, err)
		return
	}
	if remaining != 0 && window != 0 {
		// We postpone subsequent requests by at least win/rem
		msDelay := int64(window/time.Millisecond) / remaining
		delay := time.Duration(msDelay) * time.Millisecond
		c.blockReqsBefore = time.Now().Add(delay)
	}

	// Success response
	if 200 <= resp.StatusCode && resp.StatusCode <= 299 {
		return
	}

	// Non-success HTTP status codes with error body
	//err, ok := httpStatusToError[resp.StatusCode]

	// Error JSON in LogicMonitor response
	lmerr := &LMError{}
	err = json.Unmarshal(respBody, lmerr)
	if err == nil {
		err = fmt.Errorf("request %v: %w", uri, lmerr)
		return
	}

	/* Non-standard HTTP status codes

	This may not be necessary, but I can't tell from the
	documentation alone, which claims they "use standard HTTP
	error codes to indicate the error in the response body."
	This may be removed at some point.
	*/
	err, ok := lmStatusToError[resp.StatusCode]
	if ok {
		return
	}

	/* Non-success HTTP status codes without error body */
	err, ok = httpStatusToError[resp.StatusCode]
	if ok {
		return
	}
	panic(fmt.Sprintf("BUG: unexpected response: %v", string(respBody)))
}

func (c *LMClient) Get(resource url.URL) ([]byte, error) {
	return c.request(http.MethodGet, &resource, nil)
}

type pagedURI struct {
	uri *url.URL
}

func (u pagedURI) PageURI(limit, offset uint64) url.URL {
	if limit > MaxPageSize {
		limit = MaxPageSize
	}
	uri := *u.uri
	q := (&uri).Query()
	q.Add("offset", strconv.FormatUint(offset, 10))
	q.Add("size", strconv.FormatUint(limit, 10))
	(&uri).RawQuery = q.Encode()
	return uri
}

func NewLMAggregate[T any]() *LMAggregate[T] {
	return &LMAggregate[T]{Items: make([]T, 0, 64)}
}

type LMAggregate[T any] struct {
	Total int32 `json:"total"`
	Items []T   `json:"items"`
}

func (a *LMAggregate[_]) Count() uint64 {
	return uint64(a.Total)
}

func (a *LMAggregate[T]) Elems() []T {
	return a.Items
}

func makeLMPager[T any](
	client *LMClient,
	resource *url.URL,
) dp.Pager[T] {
	return dp.NewPager(
		pagedURI{resource},
		client,
		MaxPageSize,
		func() dp.Page[T] {
			return NewLMAggregate[T]()
		},
	)
}

type LMGraphLine struct {
	Legend string `json:"legend"` // "192.0.2.1:0"
	Type   string `json:"type"`   // "Stack"
	Data   []float32
}

type LMTopTalkersGraph struct {
	Type          string `json:"type"` // "graph"
	Id            int32  `json:"id"`
	Name          string `json:"name"`  // "total"
	Title         string `json:"title"` // "Throughput"
	VerticalLabel string `json:"verticalLabel"`
	Rigid         bool   `json:"rigid"` // false
	Width         int32  `json:"width"`
	Height        int32  `json:"height"`
	DisplayPrio   int32  `json:"displayPrio"`
	TimeScale     string `json:"timeScale"`
	Base1024      bool   `json:"base1024"`
	//MaxValue int32 `json:"maxValue"`  may return NaN
	//MinValue int32 `json:"minValue"`  may return NaN
	StartTime      int64       `json:"startTime"`
	EndTime        int64       `json:"endTime"`
	TimeZone       string      `json:"timeZone"`   // "CDT"
	TimeZoneId     string      `json:"timeZoneId"` // America/Chicago
	StartTZOffset  int32       `json:"startTZOffset"`
	EndTZOffset    int32       `json:"endTZOffset"`
	Step           int32       `json:"step"`           // 0
	DsName         string      `json:"dsName"`         // null
	XAxisName      string      `json:"xAxisName"`      // "2023-08-08 06:30 to 2023-08-08 19:30 CDT"
	Base           int32       `json:"base"`           // 1000
	ExportFileName string      `json:"exportFileName"` // "Adams2120-RTR-jds-2023-08-08_06:30_to_2023-08-08_19:30_CDT"
	Scopes         []*struct { // probably don't need this
		Type string `json:"type"` // "device"
		Id   int32  `json:"id"`
	} `json:"scopes"`
	// Instances []?  not sure what this is
	Timestamps []int64        `json:"timestamps"`
	Lines      []*LMGraphLine `json:"lines"`
}

type LMFlowDirection string

const (
	LMFlowDirectionIngress LMFlowDirection = "ingress"
	LMFlowDirectionEgress  LMFlowDirection = "egress"
	LMFlowDirectionBoth    LMFlowDirection = "both"
)

type LMProtocol string

const (
	LMProtocolGRE LMProtocol = "gre"
)

type LMTopCount int32

const (
	LMTopCount10  LMTopCount = 10
	LMTopCount20  LMTopCount = 20
	LMTopCount50  LMTopCount = 50
	LMTopCount100 LMTopCount = 100
)

type LMIPVersion string

const (
	LMIPVersion4    LMIPVersion = "ipv4"
	LMIPVersion6    LMIPVersion = "ipv6"
	LMIPVersionBoth LMIPVersion = "both"
)

type LMNetflowFilter struct {
	Direction LMFlowDirection `json:"direction"`
	IPVersion LMIPVersion     `json:"ipVersion"`
	Protocol  LMProtocol      `json:"protocol"`
	Top       LMTopCount      `json:"top"`
}

func (f LMNetflowFilter) String() string {
	bytes, err := json.Marshal(f)
	if err != nil {
		panic(fmt.Sprintf("BUG: failed to marshal lmnetflowfilter: %v", err))
	}
	return string(bytes)
}

func (c *LMClient) GetTopTalkersGraph(
	id int32,
	start,
	end int64,
	netflowFilter LMNetflowFilter,
) (graph *LMTopTalkersGraph, err error) {
	resStr :=
		fmt.Sprintf("/device/devices/%d/topTalkersGraph", id)

	resource, err := url.Parse(resStr)
	if err != nil {
		panic(fmt.Sprintf("BUG: get top talkers graph: failed to parse uri path '%s': %+v", resStr, err))
	}

	query := resource.Query()
	query.Add("netflowFilter", netflowFilter.String())
	query.Add("time", "zoom")
	query.Add("start", strconv.FormatInt(start, 10))
	query.Add("end", strconv.FormatInt(end, 10))
	resource.RawQuery = query.Encode()

	body, err := c.Get(*resource)
	if err != nil {
		return
	}

	graph = &LMTopTalkersGraph{}
	err = json.Unmarshal(body, graph)
	if err != nil {
		err = fmt.Errorf("get top talkers graph: %v", err)
		return
	}
	return
}

type LMWidget struct {
	Id             int32  `json:"id"`
	Type           string `json:"type"`
	Name           string `json:"name"`
	Description    string `json:"description"`
	Timescale      string `json:"timescale"`
	Interval       int32  `json:"interval"`
	LastUpdateOn   int64  `json:"lastUpdatedOn"`
	LastUpdateBy   string `json:"lastUpdatedBy"`
	UserPermission string `json:"userPermission"`
	DashboardId    int32  `json:"dashboardId"`
	Theme          string `json:"theme"`
	//NetflowFilter     string `json:"netflowFilter"`
	DeviceId          int32  `json:"deviceId"`
	Graph             string `json:"graph"`
	DeviceDisplayName string `json:"deviceDisplayName"`
}

// Get widgets
func (c *LMClient) GetWidgetList() dp.Pager[*LMWidget] {
	resStr := "/dashboard/widgets"
	resource, err := url.Parse(resStr)
	if err != nil {
		panic(fmt.Sprintf("BUG: get widget list: failed to parse uri path '%s': %+v", resStr, err))
	}
	return makeLMPager[*LMWidget](c, resource)
}

type LMDevice struct {
	Id                   int32  `json:"id"`
	Name                 string `json:"name"`
	DisplayName          string `json:"displayName"`
	PreferredCollectorId int32  `json:"preferredCollectorId"`
}

// Get devices by group id
func (
	c *LMClient,
) GetImmediateDeviceList(
	id int32,
) dp.Pager[*LMDevice] {
	resStr := fmt.Sprintf("/device/groups/%d/devices", id)
	resource, err := url.Parse(resStr)
	if err != nil {
		panic(fmt.Sprintf("BUG: get widget list: failed to parse uri path '%s': %+v", resStr, err))
	}
	return makeLMPager[*LMDevice](c, resource)
}

type LMNetflowFlow struct {
	DataType        string  `json:"dataType"`
	FirstEpochInSec int64   `json:"firstEpochInSec"`
	LastEpochInSec  int64   `json:"lastEpochInSec"`
	Usage           float32 `json:"usage"`
	PercentUsage    float32 `json:"percentUsage"`
	IfIn            int64   `json:"ifIn"`
	IfOut           int64   `json:"ifOut"`
	SrcASN          int64   `json:"srcASN"`
	SrcAsnName      string  `json:"srcAsnName"`
	DstASN          int64   `json:"dstASN"`
	DstAsnName      string  `json:"destAsnName"`
	SrcIP           string  `json:"srcIP"`
	DstIP           string  `json:"dstIP"`
	Protocol        string  `json:"protocol"`
	SrcPort         int32   `json:"srcPort"`
	DstPort         int32   `json:"dstPort"`
	SrcDNS          string  `json:"srcDNS"`
	DstDNS          string  `json:"dstDNS"`
	SrcMBytes       float32 `json:"sourceMBytes"`
	DstMBytes       float32 `json:"destinationMBytes"`
}

// Get Netflow flows by device id, date range, and Netflow
// filter
func (
	c *LMClient,
) GetNetflowFlowList(
	id int32,
	start,
	end int64,
	netflowFilter string,
) dp.Pager[*LMNetflowFlow] {
	resStr := fmt.Sprintf("/device/devices/%d/flows", id)
	resource, err := url.Parse(resStr)
	if err != nil {
		panic(fmt.Sprintf("BUG: get widget list: failed to parse uri path '%s': %+v", resStr, err))
	}

	query := resource.Query()
	query.Add("netflowFilter", netflowFilter)
	query.Add("start", strconv.FormatInt(start, 10))
	query.Add("end", strconv.FormatInt(end, 10))
	resource.RawQuery = query.Encode()

	return makeLMPager[*LMNetflowFlow](c, resource)
}