Peernet-Archive/Cmd
1
0
Fork 0
mirror of https://github.com/PeernetOfficial/Cmd.git synced 2026-07-17 02:47:52 +01:00
Code Issues Packages Projects Releases Wiki Activity

Add high-level filters for incoming and outgoing messages

Browse Source
This commit is contained in:
Kleissner
2021-07-31 17:36:29 +02:00
parent d9573cdb15
commit 1ff5797d3c
5 changed files with 251 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

242
Command Debug.go
View File

@@ -7,6 +7,7 @@ Author: Peter Kleissner
package main
import (
"bytes"
"encoding/hex"
"fmt"
"sync"
@@ -14,6 +15,7 @@ import (
"github.com/PeernetOfficial/core"
"github.com/PeernetOfficial/core/dht"
"github.com/btcsuite/btcd/btcec"
)
// debugCmdConnect connects to the node ID
@@ -28,8 +30,8 @@ func debugCmdConnect(nodeID []byte) {
} else {
fmt.Printf("* In local routing table: No. Lookup via DHT. Timeout = 10 seconds.\n")
addKeyMonitor(nodeID)
defer removeKeyMonitor(nodeID)
hashMonitorControl(nodeID, 0)
defer hashMonitorControl(nodeID, 1)
// Discovery via DHT.
_, peer, _ = core.FindNode(nodeID, time.Second*10)
@@ -61,27 +63,39 @@ var monitorKeys map[string]struct{}
var monitorKeysMutex sync.RWMutex
var enableMonitorAll = false // Enables output for all searches. Otherwise it only monitors searches stored in monitorKeys.
func addKeyMonitor(key []byte) {
// hashMonitorControl adds (0), removes (1), or inverts (2) a hash on the list
func hashMonitorControl(key []byte, action int) (added bool) {
monitorKeysMutex.Lock()
monitorKeys[string(key)] = struct{}{}
monitorKeysMutex.Unlock()
defer monitorKeysMutex.Unlock()
switch action {
case 0:
monitorKeys[string(key)] = struct{}{}
added = true
case 1:
delete(monitorKeys, string(key))
case 2:
if _, ok := monitorKeys[string(key)]; !ok {
monitorKeys[string(key)] = struct{}{}
added = true
} else {
delete(monitorKeys, string(key))
}
}
return
}
func removeKeyMonitor(key []byte) {
func hashIsMonitored(key []byte) (monitored bool) {
monitorKeysMutex.Lock()
delete(monitorKeys, string(key))
_, monitored = monitorKeys[string(key)]
monitorKeysMutex.Unlock()
return
}
func filterSearchStatus(client *dht.SearchClient, function, format string, v ...interface{}) {
// check if the search client is actively being monitored
if !enableMonitorAll {
monitorKeysMutex.Lock()
_, ok := monitorKeys[string(client.Key)]
monitorKeysMutex.Unlock()
if !ok {
return
}
if !enableMonitorAll && !hashIsMonitored(client.Key) {
return
}
keyA := client.Key
@@ -108,7 +122,7 @@ func filterSearchStatus(client *dht.SearchClient, function, format string, v ...
var enableWatchIncomingAll = false
func filterIncomingRequest(peer *core.PeerInfo, Action int, Key []byte, Info interface{}) {
if !enableWatchIncomingAll {
if !enableWatchIncomingAll && !hashIsMonitored(peer.NodeID) {
return
}
@@ -124,5 +138,199 @@ func filterIncomingRequest(peer *core.PeerInfo, Action int, Key []byte, Info int
requestType = "INFO_STORE"
}
fmt.Printf("Incoming info request %s from %s for key %s\n", requestType, hex.EncodeToString(peer.NodeID), hex.EncodeToString(Key))
if Action == core.ActionFindSelf && bytes.Equal(peer.NodeID, Key) {
fmt.Printf("Info request from %s %s\n", hex.EncodeToString(peer.NodeID), requestType)
} else {
fmt.Printf("Info request from %s %s for key %s\n", hex.EncodeToString(peer.NodeID), requestType, hex.EncodeToString(Key))
}
}
// ---- filter for incoming and outgoing packets ----
func filterMessageIn(peer *core.PeerInfo, raw *core.MessageRaw, message interface{}) {
if !hashIsMonitored(peer.NodeID) {
// TODO: For Announcement/Response also check data, Traverse the final target
return
}
commandA := "Unknown"
switch raw.Command {
case core.CommandAnnouncement:
commandA = "Announcement"
case core.CommandResponse:
commandA = "Response"
case core.CommandPing:
commandA = "Ping"
case core.CommandPong:
commandA = "Pong"
case core.CommandLocalDiscovery:
commandA = "Local Discovery"
case core.CommandTraverse:
commandA = "Traverse"
case core.CommandChat:
commandA = "Chat"
}
output := fmt.Sprintf("-------- Node %s Incoming %s --------\n", hex.EncodeToString(peer.NodeID), commandA)
output += fmt.Sprintf("Sender Peer ID: %s\n", hex.EncodeToString(peer.PublicKey.SerializeCompressed()))
if !raw.SenderPublicKey.IsEqual(peer.PublicKey) {
output += fmt.Sprintf("WARNING: Mismatch of public keys, sender %s and packet indicates %s\n", hex.EncodeToString(peer.PublicKey.SerializeCompressed()), hex.EncodeToString(raw.SenderPublicKey.SerializeCompressed()))
}
if message == nil {
output += "(no message decoded)\n"
} else if announce, ok := message.(*core.MessageAnnouncement); ok {
output += fmt.Sprintf("Fields:\n Protocol supported %d\n", announce.Protocol)
output += fmt.Sprintf(" Feature bits %d\n", announce.Features)
output += fmt.Sprintf(" Action bits %d\n", announce.Actions)
output += fmt.Sprintf(" Blockchain Height %d\n", announce.BlockchainHeight)
output += fmt.Sprintf(" Blockchain Version %d\n", announce.BlockchainVersion)
output += fmt.Sprintf(" Port Internal %d\n", announce.PortInternal)
output += fmt.Sprintf(" Port External %d\n", announce.PortExternal)
output += fmt.Sprintf(" User Agent %s\n", announce.UserAgent)
if len(announce.FindPeerKeys) > 0 {
output += fmt.Sprintf("FIND_PEER %d records:\n", len(announce.FindPeerKeys))
}
for _, find := range announce.FindPeerKeys {
output += fmt.Sprintf(" - Find peer %s\n", hex.EncodeToString(find.Hash))
}
if len(announce.FindDataKeys) > 0 {
output += fmt.Sprintf("FIND_VALUE %d records:\n", len(announce.FindDataKeys))
}
for _, find := range announce.FindDataKeys {
output += fmt.Sprintf(" - Find data %s\n", hex.EncodeToString(find.Hash))
}
if len(announce.InfoStoreFiles) > 0 {
output += fmt.Sprintf("INFO_STORE %d records:\n", len(announce.InfoStoreFiles))
}
for _, info := range announce.InfoStoreFiles {
output += fmt.Sprintf(" - Info store %s, type %d, size %d\n", hex.EncodeToString(info.ID.Hash), info.Type, info.Size)
}
} else if response, ok := message.(*core.MessageResponse); ok {
output += fmt.Sprintf("Fields:\n Protocol supported %d\n", response.Protocol)
output += fmt.Sprintf(" Feature bits %d\n", response.Features)
output += fmt.Sprintf(" Action bits %d\n", response.Actions)
output += fmt.Sprintf(" Blockchain Height %d\n", response.BlockchainHeight)
output += fmt.Sprintf(" Blockchain Version %d\n", response.BlockchainVersion)
output += fmt.Sprintf(" Port Internal %d\n", response.PortInternal)
output += fmt.Sprintf(" Port External %d\n", response.PortExternal)
output += fmt.Sprintf(" User Agent %s\n", response.UserAgent)
for _, hash := range response.Hash2Peers {
isLast := ""
if hash.IsLast {
isLast = " [last result in sequence]"
}
output += fmt.Sprintf(" - Peers known for the hash %s%s\n", hex.EncodeToString(hash.ID.Hash), isLast)
for n := range hash.Closest {
output += fmt.Sprintf(" Close peer:\n%s\n", outputPeerRecord(&hash.Closest[n]))
}
for n := range hash.Storing {
output += fmt.Sprintf(" Peer stores:\n%s\n", outputPeerRecord(&hash.Storing[n]))
}
}
for _, find := range response.FilesEmbed {
output += fmt.Sprintf(" - File embedded %s (%d bytes)\n", hex.EncodeToString(find.ID.Hash), len(find.Data))
}
for _, hash := range response.HashesNotFound {
output += fmt.Sprintf(" - Hash not found %s\n", hex.EncodeToString(hash))
}
} else if traverse, ok := message.(*core.MessageTraverse); ok {
output += fmt.Sprintf("Fields:\n Target Peer %s\n", hex.EncodeToString(traverse.TargetPeer.SerializeCompressed()))
output += fmt.Sprintf(" Authorized Relay Peer %s\n", hex.EncodeToString(traverse.AuthorizedRelayPeer.SerializeCompressed()))
output += fmt.Sprintf(" Signer Public Key %s\n", hex.EncodeToString(traverse.SignerPublicKey.SerializeCompressed()))
output += fmt.Sprintf(" Expires %s\n", traverse.Expires.String())
output += fmt.Sprintf(" IPv4 %s\n", traverse.IPv4.String())
output += fmt.Sprintf(" Port IPv4 %d\n", traverse.PortIPv4)
output += fmt.Sprintf(" Port IPv4 Reported External %d\n", traverse.PortIPv4ReportedExternal)
output += fmt.Sprintf(" IPv6 %s\n", traverse.IPv6.String())
output += fmt.Sprintf(" Port IPv6 %d\n", traverse.PortIPv6)
output += fmt.Sprintf(" Port IPv6 Reported External %d\n", traverse.PortIPv6ReportedExternal)
}
output += "--------\n"
fmt.Printf("%s", output)
}
func outputPeerRecord(record *core.PeerRecord) (output string) {
output += fmt.Sprintf(" * Peer ID %s\n", hex.EncodeToString(record.PublicKey.SerializeCompressed()))
output += fmt.Sprintf(" Node ID %s\n", hex.EncodeToString(record.NodeID))
if record.IPv4 != nil && !record.IPv4.IsUnspecified() {
output += fmt.Sprintf(" IPv4 %s\n", record.IPv4.String())
output += fmt.Sprintf(" Port IPv4 %d\n", record.IPv4Port)
output += fmt.Sprintf(" Port IPv4 Reported Internal %d\n", record.IPv4PortReportedInternal)
output += fmt.Sprintf(" Port IPv4 Reported External %d\n", record.IPv4PortReportedExternal)
}
if record.IPv6 != nil && !record.IPv6.IsUnspecified() {
output += fmt.Sprintf(" IPv6 %s\n", record.IPv6.String())
output += fmt.Sprintf(" Port IPv6 %d\n", record.IPv6Port)
output += fmt.Sprintf(" Port IPv6 Reported Internal %d\n", record.IPv6PortReportedInternal)
output += fmt.Sprintf(" Port IPv6 Reported External %d\n", record.IPv6PortReportedExternal)
}
output += fmt.Sprintf(" Last Contact %s", record.LastContactT.Format(dateFormat))
return
}
func outputOutgoingMessage(peer *core.PeerInfo, packet *core.PacketRaw) {
if !hashIsMonitored(peer.NodeID) {
// TODO: For Announcement/Response also check data, Traverse the final target
return
}
commandA := "Unknown"
switch packet.Command {
case core.CommandAnnouncement:
commandA = "Announcement"
case core.CommandResponse:
commandA = "Response"
case core.CommandPing:
commandA = "Ping"
case core.CommandPong:
commandA = "Pong"
case core.CommandLocalDiscovery:
commandA = "Local Discovery"
case core.CommandTraverse:
commandA = "Traverse"
case core.CommandChat:
commandA = "Chat"
}
output := fmt.Sprintf("-------- Node %s Outgoing %s --------\n", hex.EncodeToString(peer.NodeID), commandA)
output += fmt.Sprintf("Receiver Peer ID: %s\n", hex.EncodeToString(peer.PublicKey.SerializeCompressed()))
// TODO: Decoding of payload data (done by caller of this function)
output += "--------\n"
fmt.Printf("%s", output)
}
func filterMessageOutAnnouncement(receiverPublicKey *btcec.PublicKey, peer *core.PeerInfo, packet *core.PacketRaw, findSelf bool, findPeer []core.KeyHash, findValue []core.KeyHash, files []core.InfoStore) {
if peer == nil {
peer = &core.PeerInfo{PublicKey: receiverPublicKey, NodeID: core.PublicKey2NodeID(receiverPublicKey)}
}
outputOutgoingMessage(peer, packet)
}
func filterMessageOutResponse(peer *core.PeerInfo, packet *core.PacketRaw, hash2Peers []core.Hash2Peer, filesEmbed []core.EmbeddedFileData, hashesNotFound [][]byte) {
outputOutgoingMessage(peer, packet)
}
func filterMessageOutTraverse(peer *core.PeerInfo, packet *core.PacketRaw, embeddedPacket *core.PacketRaw, receiverEnd *btcec.PublicKey) {
outputOutgoingMessage(peer, packet)
}
func filterMessageOutPing(peer *core.PeerInfo, packet *core.PacketRaw, connection *core.Connection) {
outputOutgoingMessage(peer, packet)
}
func filterMessageOutPong(peer *core.PeerInfo, packet *core.PacketRaw) {
outputOutgoingMessage(peer, packet)
}

17
Command Line.go
View File

@@ -94,6 +94,7 @@ func showHelp() {
"debug connect Attempts to connect to the target peer\n" +
"debug watch searches Watch all outgoing DHT searches\n" +
"debug watch incoming Watch all incoming information requests\n" +
"debug watch Watch packets and info requests for hash\n" +
"hash Create blake3 hash of input\n" +
"warehouse get Get data from local warehouse by hash\n" +
"warehouse store Store data into local warehouse\n" +
@@ -366,6 +367,22 @@ func userCommands() {
dht.DisableBucketRefresh = number == 1
}
case "debug watch":
fmt.Printf("Enter hash of data or node ID to watch. This monitors info requests and packets. Enter same hash again to remove from list.\n")
text, _ := getUserOptionString(reader)
var hash []byte
var err error
if hash, err = hex.DecodeString(text); err != nil || len(hash) != 256/8 {
fmt.Printf("Invalid hash. Hex-encoded 64 character hash expected.\n")
break
}
added := hashMonitorControl(hash, 2)
if added {
fmt.Printf("The hash was added to the monitoring list.\n")
} else {
fmt.Printf("The hash was removed from the monitoring list.\n")
}
}
}
}

6
Main.go
View File

@@ -47,6 +47,12 @@ func init() {
core.Filters.LogError = logError
core.Filters.DHTSearchStatus = filterSearchStatus
core.Filters.IncomingRequest = filterIncomingRequest
core.Filters.MessageIn = filterMessageIn
core.Filters.MessageOutAnnouncement = filterMessageOutAnnouncement
core.Filters.MessageOutResponse = filterMessageOutResponse
core.Filters.MessageOutTraverse = filterMessageOutTraverse
core.Filters.MessageOutPing = filterMessageOutPing
core.Filters.MessageOutPong = filterMessageOutPong
core.Init()
}

2
go.mod
View File

@@ -3,6 +3,6 @@ module github.com/PeernetOfficial/Cmd
go 1.16
require (
github.com/PeernetOfficial/core v0.0.0-20210729021053-3233bf5c1bf6
github.com/PeernetOfficial/core v0.0.0-20210731152914-2937154c8d7b
github.com/btcsuite/btcd v0.22.0-beta.0.20210625194946-86a17263b0ff
)

4
go.sum
View File

@@ -1,5 +1,5 @@
github.com/PeernetOfficial/core v0.0.0-20210729021053-3233bf5c1bf6 h1:J4Tb6sP7fnOrTsrJtXaSDqA1CpEYPmF6lg6QjeOUm0I=
github.com/PeernetOfficial/core v0.0.0-20210729021053-3233bf5c1bf6/go.mod h1:9aIe6MS6tefJnfmyEZFqaNF4NIeq7wJutV7EU+uowps=
github.com/PeernetOfficial/core v0.0.0-20210731152914-2937154c8d7b h1:y/Afz6M1vlLpn5nmZmHyK0Tze7hOl9N7p43HfTb578I=
github.com/PeernetOfficial/core v0.0.0-20210731152914-2937154c8d7b/go.mod h1:9aIe6MS6tefJnfmyEZFqaNF4NIeq7wJutV7EU+uowps=
github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII=
github.com/btcsuite/btcd v0.20.1-beta/go.mod h1:wVuoA8VJLEcwgqHBwHmzLRazpKxTv13Px/pDuV7OomQ=
github.com/btcsuite/btcd v0.21.0-beta.0.20210401013323-36a96f6a0025/go.mod h1:9n5ntfhhHQBIhUvlhDvD3Qg6fRUj4jkN0VB8L8svzOA=