Add configurable HTTP verb and update BlueFlow field names

api.go

@@ -2,11 +2,12 @@
 Submit information about discovered devices to other services via REST API
 endpoints.
 
-Upon discovering and/or identifying a device, submit a POST request to a URL
-like https://my-other-system.com/device/, optionally presenting an authorization
-token.  This API endpoint should behave like an "upsert" request, i.e., it
-should update itself rather than bail when it receives a clue about a device it
-has been told about before.
+Upon discovering and/or identifying a device, submit a request to a URL like
+https://my-other-system.com/device/, optionally presenting an authorization
+token.  The HTTP method is configurable (PUT by default, POST for legacy
+endpoints).  This API endpoint should behave like an "upsert" request, i.e.,
+it should update itself rather than bail when it receives a clue about a device
+it has been told about before.
 
 Concurrent requests to this API endpoint are limited according to the apiLimit
 parameter of NewAPIClient(). If apiLimit number of requests are currently in
@@ -31,25 +32,28 @@ import (
 // An APIClient holds state and credentials related to uploading Asset
 // information to a REST API endpoint.
 type APIClient struct {
-	url       string
-	authToken string
-	clientID  string
-	enabled   bool
-	semaphore chan bool
+	url        string
+	authToken  string
+	clientID   string
+	httpMethod string
+	enabled    bool
+	semaphore  chan bool
 }
 
 // NewAPIClient creates a new APIClient.
 func NewAPIClient(
 	apiURL string,
 	apiToken string,
 	clientID string,
+	httpMethod string,
 	apiLimit int,
 	enabled bool,
 ) *APIClient {
 	apiClient := new(APIClient)
 	apiClient.url = apiURL
 	apiClient.authToken = apiToken
 	apiClient.clientID = clientID
+	apiClient.httpMethod = httpMethod
 	apiClient.enabled = enabled
 
 	// A channel will act as a semaphore with the desired level of concurrency
@@ -85,7 +89,7 @@ func (apiClient *APIClient) Upload(asset *Asset) (map[string]interface{}, error)
 		Timeout: time.Duration(5 * time.Second),
 	}
 	request, err := http.NewRequest(
-		http.MethodPost,
+		apiClient.httpMethod,
 		apiClient.url,
 		bytes.NewBuffer(bytesRepresentation),
 	)

api_test.go

@@ -32,7 +32,7 @@ func setup() func() {
 	server = httptest.NewServer(mux)
 
 	apiURL := server.URL + "/api" // Base URL automatically chosen by httptest
-	apiClient = NewAPIClient(apiURL, "", "", 1, true)
+	apiClient = NewAPIClient(apiURL, "", "", "PUT", 1, true)
 	stats = *NewStats()
 
 	// The API client calls the logger global.  Initialize it.
@@ -62,7 +62,7 @@ func TestAPISimple(t *testing.T) {
 	result, err := apiClient.Upload(&Asset{
 		"10.0.0.1",
 		"0000:0000:0000:0000:0000:FFFF:0A00:0001",
-		"8000",
+		[]int{8000},
 		"2575",
 		"11:22:33:44:55:66",
 		"Hospira Plum A+",

asset.go

@@ -2,21 +2,31 @@ package main
 
 import (
 	"encoding/csv"
+	"fmt"
 	"os"
+	"strings"
 	"sync"
 	"time"
 )
 
+// NewAsset creates an Asset with safe defaults (e.g., empty slice instead of nil
+// for ListensOnPorts so it serializes as [] rather than null in JSON).
+func NewAsset() *Asset {
+	return &Asset{
+		ListensOnPorts: []int{},
+	}
+}
+
 // An Asset represents an observation of one endpoint seen in network traffic.
 //
 // Each field is annotated with its JSON field name.
 type Asset struct {
-	IPv4Address    string    `json:"ipv4_address"`
+	IPv4Address    string    `json:"ip_address"`
 	IPv6Address    string    `json:"ipv6_address"`
-	ListensOnPort  string    `json:"open_port_tcp"`
+	ListensOnPorts []int     `json:"open_ports_tcp"`
 	ConnectsToPort string    `json:"connect_port_tcp"`
 	MACAddress     string    `json:"mac_address"`
-	Identifier     string    `json:"identifier"`
+	Identifier     string    `json:"name"`
 	Provenance     string    `json:"provenance"`
 	LastSeen       time.Time `json:"last_seen"`
 	ClientID       string    `json:"client_id"`
@@ -54,12 +64,12 @@ func NewAssetCSVWriter(filename string) (*AssetCSVWriter, error) {
 
 	// Write CSV header
 	header := []string{
-		"ipv4_address",
+		"ip_address",
 		"ipv6_address",
-		"open_port_tcp",
+		"open_ports_tcp",
 		"connect_port_tcp",
 		"mac_address",
-		"identifier",
+		"name",
 		"provenance",
 		"last_seen",
 		"client_id",
@@ -94,9 +104,14 @@ func (w *AssetCSVWriter) Append(asset *Asset) error {
 	defer w.Unlock()
 
 	// Write CSV row
+	ports := make([]string, len(asset.ListensOnPorts))
+	for i, p := range asset.ListensOnPorts {
+		ports[i] = fmt.Sprintf("%d", p)
+	}
 	row := []string{
 		asset.IPv4Address,
 		asset.IPv6Address,
+		strings.Join(ports, ";"),
 		asset.ConnectsToPort,
 		asset.MACAddress,
 		asset.Identifier,

asset_test.go

@@ -18,7 +18,7 @@ func TestAssetCSV(t *testing.T) {
 	asset := &Asset{
 		"10.0.0.1",
 		"0000:0000:0000:0000:0000:FFFF:0A00:0001",
-		"8000",
+		[]int{8000},
 		"2575",
 		"11:22:33:44:55:66",
 		"Hospira Plum A+",
@@ -42,9 +42,9 @@ func TestAssetCSV(t *testing.T) {
 	if err != nil {
 		panic(err)
 	}
-	expected := `ipv4_address,ipv6_address,open_port_tcp,connect_port_tcp,mac_address,identifier,provenance,last_seen,client_id
-10.0.0.1,0000:0000:0000:0000:0000:FFFF:0A00:0001,2575,11:22:33:44:55:66,Hospira Plum A+,HL7,0001-01-01 00:00:00 +0000 UTC,ID0
-10.0.0.1,0000:0000:0000:0000:0000:FFFF:0A00:0001,2575,11:22:33:44:55:66,Hospira Plum A+,HL7,0001-01-01 00:00:00 +0000 UTC,ID0
+	expected := `ip_address,ipv6_address,open_ports_tcp,connect_port_tcp,mac_address,name,provenance,last_seen,client_id
+10.0.0.1,0000:0000:0000:0000:0000:FFFF:0A00:0001,8000,2575,11:22:33:44:55:66,Hospira Plum A+,HL7,0001-01-01 00:00:00 +0000 UTC,ID0
+10.0.0.1,0000:0000:0000:0000:0000:FFFF:0A00:0001,8000,2575,11:22:33:44:55:66,Hospira Plum A+,HL7,0001-01-01 00:00:00 +0000 UTC,ID0
 `
 	if string(actual) != expected {
 		t.Errorf("CSV file actual %s does not match expected: %s\n", actual, expected)

go.mod

@@ -0,0 +1,15 @@
+module github.com/virtalabs/tapirx
+
+go 1.26
+
+require (
+	github.com/google/gopacket v1.1.19
+	github.com/virtalabs/hl7 v0.0.0-20181026220847-0785fb8f7db4
+)
+
+require (
+	github.com/facebookgo/stack v0.0.0-20160209184415-751773369052 // indirect
+	github.com/facebookgo/stackerr v0.0.0-20150612192056-c2fcf88613f4 // indirect
+	github.com/stretchr/testify v1.11.1 // indirect
+	golang.org/x/sys v0.0.0-20190412213103-97732733099d // indirect
+)

go.sum

@@ -0,0 +1,30 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/facebookgo/stack v0.0.0-20160209184415-751773369052 h1:JWuenKqqX8nojtoVVWjGfOF9635RETekkoH6Cc9SX0A=
+github.com/facebookgo/stack v0.0.0-20160209184415-751773369052/go.mod h1:UbMTZqLaRiH3MsBH8va0n7s1pQYcu3uTb8G4tygF4Zg=
+github.com/facebookgo/stackerr v0.0.0-20150612192056-c2fcf88613f4 h1:fP04zlkPjAGpsduG7xN3rRkxjAqkJaIQnnkNYYw/pAk=
+github.com/facebookgo/stackerr v0.0.0-20150612192056-c2fcf88613f4/go.mod h1:SBHk9aNQtiw4R4bEuzHjVmZikkUKCnO1v3lPQ21HZGk=
+github.com/google/gopacket v1.1.19 h1:ves8RnFZPGiFnTS0uPQStjwru6uO6h+nlr9j6fL7kF8=
+github.com/google/gopacket v1.1.19/go.mod h1:iJ8V8n6KS+z2U1A8pUwu8bW5SyEMkXJB8Yo/Vo+TKTo=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+github.com/virtalabs/hl7 v0.0.0-20181026220847-0785fb8f7db4 h1:pK1pnBbGNR1tSc9ePrPWVT2J2wxCDsQgo0BFnKHPAkg=
+github.com/virtalabs/hl7 v0.0.0-20181026220847-0785fb8f7db4/go.mod h1:Pya115JrQc1Z7A77wKL7njd1A9kO5DLELfF91FC+EMA=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
+golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
+golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859 h1:R/3boaszxrf1GEUWTVDzSKVwLmSJpwZ1yqXm8j0v2QI=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190412213103-97732733099d h1:+R4KGOnez64A81RvjARKc4UT5/tI9ujCIVX+P5KiHuI=
+golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
+golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

hl7_decode_test.go

@@ -108,7 +108,7 @@ func getNRecordString(nrec int) string {
 	}
 	alphas := make([]string, nrec)
 	for i := 0; i < nrec; i++ {
-		alphas[i] = string('A' + i)
+		alphas[i] = string(rune('A' + i))
 	}
 	return strings.Join(alphas, "|")
 }

main.go

@@ -120,10 +120,16 @@ func main() {
 	version := flag.Bool("version", false, "Show version information and exit")
 	packetLimit := flag.Int("limit", 0, "Exit after N packets, 0 for unlimited")
 	sequential := flag.Bool("sequential", false, "Process packets sequentially")
+	httpVerb := flag.String("httpverb", "PUT", "HTTP method for API upsert (PUT or POST)")
 	csvFilename := flag.String("csv", "", "Stream assets to CSV file")
 	listIfaces := flag.Bool("interfaces", false, "List all network interfaces and exit")
 	flag.Parse()
 
+	if *httpVerb != "PUT" && *httpVerb != "POST" {
+		fmt.Fprintf(os.Stderr, "Invalid -httpverb %q: must be PUT or POST\n", *httpVerb)
+		os.Exit(1)
+	}
+
 	setupLogging(*debug)
 	stats = *NewStats()
 
@@ -170,7 +176,7 @@ func main() {
 
 	// Configure the API client module
 	apiClientEnabled := *apiURL != ""
-	apiClient := NewAPIClient(*apiURL, *apiToken, *clientID, *apiLimit, apiClientEnabled)
+	apiClient := NewAPIClient(*apiURL, *apiToken, *clientID, *httpVerb, *apiLimit, apiClientEnabled)
 
 	// Configure CSV writer module
 	assetCSVWriter, err := NewAssetCSVWriter(*csvFilename)

packet.go

@@ -58,7 +58,7 @@ func decodeLayers(packet gopacket.Packet, asset *Asset) error {
 				// then it is *initiating* a connection to the *destination*
 				// port (i.e., it's the "client" side of a new connection).
 				if tcp.ACK {
-					asset.ListensOnPort = tcp.SrcPort.String()
+					asset.ListensOnPorts = []int{int(tcp.SrcPort)}
 					asset.Provenance = "TCP handshake"
 					stats.AddLayer("TCP/handshake")
 					logger.Printf("  TCP server on %s\n", tcp.SrcPort)
@@ -121,7 +121,7 @@ func handlePacket(
 	}
 
 	// Initialize an empty Asset to store information learned during dissection
-	asset := &Asset{}
+	asset := NewAsset()
 	asset.LastSeen = time.Now()
 
 	// Decode packet and update statistics
@@ -157,7 +157,9 @@ func handlePacket(
 	// Upload to API if requested by the user.  If the user did not specify a
 	// URL with a command line flag, the URL will be empty.
 	if apiClient.enabled {
-		if _, err := apiClient.Upload(asset); err != nil {
+		if asset.MACAddress == "" {
+			logger.Println("Skipping API upload: no MAC address")
+		} else if _, err := apiClient.Upload(asset); err != nil {
 			logger.Println("API Upload error:", err)
 			stats.AddUploadError(err)
 		} else {

packet_test.go

@@ -33,7 +33,7 @@ func TestPacketParseSimple(t *testing.T) {
 
 	// Initialize objects later used by handlePacket
 	stats = *NewStats()
-	apiClient := NewAPIClient("", "", "", 1, false)
+	apiClient := NewAPIClient("", "", "", "PUT", 1, false)
 	assetCSVWriter, err := NewAssetCSVWriter("")
 	if err != nil {
 		panic(err)

stats.go

@@ -7,6 +7,7 @@ package main
 
 import (
 	"encoding/json"
+	"fmt"
 	"sort"
 	"sync"
 )
@@ -74,8 +75,8 @@ func (s *Stats) AddAsset(asset *Asset) {
 	if asset.IPv6Address != "" {
 		s.IPv6Addresses[asset.IPv6Address]++
 	}
-	if asset.ListensOnPort != "" {
-		s.Ports[asset.ListensOnPort]++
+	for _, port := range asset.ListensOnPorts {
+		s.Ports[fmt.Sprintf("%d", port)]++
 	}
 	if asset.ConnectsToPort != "" {
 		s.Ports[asset.ConnectsToPort]++

stats_test.go

@@ -22,7 +22,7 @@ func TestStatsString(t *testing.T) {
 	stats.AddAsset(&Asset{
 		testIP,
 		"0000:0000:0000:0000:0000:FFFF:0A00:0001",
-		"8000",
+		[]int{8000},
 		"2575",
 		testMAC,
 		"Hospira Plum A+",
@@ -80,7 +80,7 @@ func TestStatsSameID(t *testing.T) {
 	stats.AddAsset(&Asset{
 		"10.0.0.1",
 		"0000:0000:0000:0000:0000:FFFF:0A00:0001",
-		"8000",
+		[]int{8000},
 		"2575",
 		"11:22:33:44:55:66",
 		"Hospira Plum A+",
@@ -92,7 +92,7 @@ func TestStatsSameID(t *testing.T) {
 	stats.AddAsset(&Asset{
 		"10.0.0.2",
 		"0000:0000:0000:0000:0000:FFFF:0A00:0002",
-		"8000",
+		[]int{8000},
 		"2575",
 		"11:22:33:44:55:67",
 		"Hospira Plum A+",
@@ -130,7 +130,7 @@ func TestStatsDifferentID(t *testing.T) {
 	stats.AddAsset(&Asset{
 		"10.0.0.1",
 		"0000:0000:0000:0000:0000:FFFF:0A00:0001",
-		"8000",
+		[]int{8000},
 		"2575",
 		"11:22:33:44:55:66",
 		"Hospira Plum A+",
@@ -142,7 +142,7 @@ func TestStatsDifferentID(t *testing.T) {
 	stats.AddAsset(&Asset{
 		"10.0.0.2",
 		"0000:0000:0000:0000:0000:FFFF:0A00:0002",
-		"9000",
+		[]int{9000},
 		"2575",
 		"11:22:33:44:55:67",
 		"Alaris 8000",
@@ -180,7 +180,7 @@ func TestStatsSameEverything(t *testing.T) {
 	stats.AddAsset(&Asset{
 		"10.0.0.1",
 		"0000:0000:0000:0000:0000:FFFF:0A00:0001",
-		"8000",
+		[]int{8000},
 		"2575",
 		"11:22:33:44:55:66",
 		"Hospira Plum A+",
@@ -192,7 +192,7 @@ func TestStatsSameEverything(t *testing.T) {
 	stats.AddAsset(&Asset{
 		"10.0.0.1",
 		"0000:0000:0000:0000:0000:FFFF:0A00:0001",
-		"8000",
+		[]int{8000},
 		"2575",
 		"11:22:33:44:55:66",
 		"Hospira Plum A+",