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

Keep track of output writer in hash monitoring functions

Browse Source
This commit is contained in:
Kleissner
2022-01-13 02:09:00 +01:00
parent 58cb870116
commit 75b5339095
4 changed files with 96 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
API.go
View File

@@ -121,7 +121,7 @@ func apiConsole(backend *core.Backend) func(w http.ResponseWriter, r *http.Reque
countRead, err := bufferW.Read(bufferW2) countRead, err := bufferW.Read(bufferW2)
if err != nil || countRead == 0 { if err != nil || countRead == 0 {
time.Sleep(250 * time.Millisecond) time.Sleep(50 * time.Millisecond)
continue continue
} }

143
Command Debug.go
View File

@@ -32,8 +32,8 @@ func debugCmdConnect(backend *core.Backend, nodeID []byte, output io.Writer) {
} else { } else {
fmt.Fprintf(output, "* In local routing table: No. Lookup via DHT. Timeout = 10 seconds.\n") fmt.Fprintf(output, "* In local routing table: No. Lookup via DHT. Timeout = 10 seconds.\n")
hashMonitorControl(nodeID, 0) hashMonitorControl(nodeID, 0, output)
defer hashMonitorControl(nodeID, 1) defer hashMonitorControl(nodeID, 1, nil)
// Discovery via DHT. // Discovery via DHT.
_, peer, _ = backend.FindNode(nodeID, time.Second*10) _, peer, _ = backend.FindNode(nodeID, time.Second*10)
@@ -68,24 +68,23 @@ func debugCmdConnect(backend *core.Backend, nodeID []byte, output io.Writer) {
// debug output of monitored keys searched in the DHT // debug output of monitored keys searched in the DHT
var monitorKeys map[string]struct{} var monitorKeys map[string]io.Writer = make(map[string]io.Writer)
var monitorKeysMutex sync.RWMutex var monitorKeysMutex sync.RWMutex
var enableMonitorAll = false // Enables output for all searches. Otherwise it only monitors searches stored in monitorKeys.
// hashMonitorControl adds (0), removes (1), or inverts (2) a hash on the list // hashMonitorControl adds (0), removes (1), or inverts (2) a hash on the list
func hashMonitorControl(key []byte, action int) (added bool) { func hashMonitorControl(key []byte, action int, output io.Writer) (added bool) {
monitorKeysMutex.Lock() monitorKeysMutex.Lock()
defer monitorKeysMutex.Unlock() defer monitorKeysMutex.Unlock()
switch action { switch action {
case 0: case 0:
monitorKeys[string(key)] = struct{}{} monitorKeys[string(key)] = output
added = true added = true
case 1: case 1:
delete(monitorKeys, string(key)) delete(monitorKeys, string(key))
case 2: case 2:
if _, ok := monitorKeys[string(key)]; !ok { if _, ok := monitorKeys[string(key)]; !ok {
monitorKeys[string(key)] = struct{}{} monitorKeys[string(key)] = output
added = true added = true
} else { } else {
delete(monitorKeys, string(key)) delete(monitorKeys, string(key))
@@ -95,15 +94,24 @@ func hashMonitorControl(key []byte, action int) (added bool) {
return return
} }
func hashIsMonitored(key []byte) (monitored bool) { func hashIsMonitored(keys ...[]byte) (monitored bool, output io.Writer) {
monitorKeysMutex.Lock() monitorKeysMutex.Lock()
_, monitored = monitorKeys[string(key)] defer monitorKeysMutex.Unlock()
monitorKeysMutex.Unlock()
return for _, key := range keys {
if output, monitored = monitorKeys[string(key)]; monitored {
return true, output
}
}
return false, nil
} }
const keyMonitorAllSearches = "all searches" // special key to monitor all searches
func filterSearchStatus(client *dht.SearchClient, function, format string, v ...interface{}) { func filterSearchStatus(client *dht.SearchClient, function, format string, v ...interface{}) {
if !enableMonitorAll && !hashIsMonitored(client.Key) { monitored, output := hashIsMonitored(client.Key, []byte(keyMonitorAllSearches))
if !monitored {
return return
} }
@@ -123,15 +131,16 @@ func filterSearchStatus(client *dht.SearchClient, function, format string, v ...
intend = " >" intend = " >"
} }
fmt.Printf(intend+" "+function+" ["+hex.EncodeToString(keyA)+"] "+format, v...) fmt.Fprintf(output, intend+" "+function+" ["+hex.EncodeToString(keyA)+"] "+format, v...)
} }
// ---- filter for incoming information requests ---- // ---- filter for incoming information requests ----
var enableWatchIncomingAll = false const keyMonitorAllRequests = "all requests" // special key to monitor all info requests
func filterIncomingRequest(peer *core.PeerInfo, Action int, Key []byte, Info interface{}) { func filterIncomingRequest(peer *core.PeerInfo, Action int, Key []byte, Info interface{}) {
if !enableWatchIncomingAll && !hashIsMonitored(peer.NodeID) { monitored, output := hashIsMonitored(peer.NodeID, []byte(keyMonitorAllRequests))
if !monitored {
return return
} }
@@ -148,16 +157,17 @@ func filterIncomingRequest(peer *core.PeerInfo, Action int, Key []byte, Info int
} }
if Action == protocol.ActionFindSelf && bytes.Equal(peer.NodeID, Key) { if Action == protocol.ActionFindSelf && bytes.Equal(peer.NodeID, Key) {
fmt.Printf("Info request from %s %s\n", hex.EncodeToString(peer.NodeID), requestType) fmt.Fprintf(output, "Info request from %s %s\n", hex.EncodeToString(peer.NodeID), requestType)
} else { } else {
fmt.Printf("Info request from %s %s for key %s\n", hex.EncodeToString(peer.NodeID), requestType, hex.EncodeToString(Key)) fmt.Fprintf(output, "Info request from %s %s for key %s\n", hex.EncodeToString(peer.NodeID), requestType, hex.EncodeToString(Key))
} }
} }
// ---- filter for incoming and outgoing packets ---- // ---- filter for incoming and outgoing packets ----
func filterMessageIn(peer *core.PeerInfo, raw *protocol.MessageRaw, message interface{}) { func filterMessageIn(peer *core.PeerInfo, raw *protocol.MessageRaw, message interface{}) {
if !hashIsMonitored(peer.NodeID) { monitored, output := hashIsMonitored(peer.NodeID)
if !monitored {
// TODO: For Announcement/Response also check data, Traverse the final target // TODO: For Announcement/Response also check data, Traverse the final target
return return
} }
@@ -181,88 +191,88 @@ func filterMessageIn(peer *core.PeerInfo, raw *protocol.MessageRaw, message inte
commandA = "Chat" commandA = "Chat"
} }
output := fmt.Sprintf("-------- Node %s Incoming %s --------\n", hex.EncodeToString(peer.NodeID), commandA) text := 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())) text += fmt.Sprintf("Sender Peer ID: %s\n", hex.EncodeToString(peer.PublicKey.SerializeCompressed()))
if !raw.SenderPublicKey.IsEqual(peer.PublicKey) { 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())) text += 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 { if message == nil {
output += "(no message decoded)\n" text += "(no message decoded)\n"
} else if announce, ok := message.(*protocol.MessageAnnouncement); ok { } else if announce, ok := message.(*protocol.MessageAnnouncement); ok {
output += fmt.Sprintf("Fields:\n Protocol supported %d\n", announce.Protocol) text += fmt.Sprintf("Fields:\n Protocol supported %d\n", announce.Protocol)
output += fmt.Sprintf(" Feature bits %d\n", announce.Features) text += fmt.Sprintf(" Feature bits %d\n", announce.Features)
output += fmt.Sprintf(" Action bits %d\n", announce.Actions) text += fmt.Sprintf(" Action bits %d\n", announce.Actions)
output += fmt.Sprintf(" Blockchain Height %d\n", announce.BlockchainHeight) text += fmt.Sprintf(" Blockchain Height %d\n", announce.BlockchainHeight)
output += fmt.Sprintf(" Blockchain Version %d\n", announce.BlockchainVersion) text += fmt.Sprintf(" Blockchain Version %d\n", announce.BlockchainVersion)
output += fmt.Sprintf(" Port Internal %d\n", announce.PortInternal) text += fmt.Sprintf(" Port Internal %d\n", announce.PortInternal)
output += fmt.Sprintf(" Port External %d\n", announce.PortExternal) text += fmt.Sprintf(" Port External %d\n", announce.PortExternal)
output += fmt.Sprintf(" User Agent %s\n", announce.UserAgent) text += fmt.Sprintf(" User Agent %s\n", announce.UserAgent)
if len(announce.FindPeerKeys) > 0 { if len(announce.FindPeerKeys) > 0 {
output += fmt.Sprintf("FIND_PEER %d records:\n", len(announce.FindPeerKeys)) text += fmt.Sprintf("FIND_PEER %d records:\n", len(announce.FindPeerKeys))
} }
for _, find := range announce.FindPeerKeys { for _, find := range announce.FindPeerKeys {
output += fmt.Sprintf(" - Find peer %s\n", hex.EncodeToString(find.Hash)) text += fmt.Sprintf(" - Find peer %s\n", hex.EncodeToString(find.Hash))
} }
if len(announce.FindDataKeys) > 0 { if len(announce.FindDataKeys) > 0 {
output += fmt.Sprintf("FIND_VALUE %d records:\n", len(announce.FindDataKeys)) text += fmt.Sprintf("FIND_VALUE %d records:\n", len(announce.FindDataKeys))
} }
for _, find := range announce.FindDataKeys { for _, find := range announce.FindDataKeys {
output += fmt.Sprintf(" - Find data %s\n", hex.EncodeToString(find.Hash)) text += fmt.Sprintf(" - Find data %s\n", hex.EncodeToString(find.Hash))
} }
if len(announce.InfoStoreFiles) > 0 { if len(announce.InfoStoreFiles) > 0 {
output += fmt.Sprintf("INFO_STORE %d records:\n", len(announce.InfoStoreFiles)) text += fmt.Sprintf("INFO_STORE %d records:\n", len(announce.InfoStoreFiles))
} }
for _, info := range 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) text += fmt.Sprintf(" - Info store %s, type %d, size %d\n", hex.EncodeToString(info.ID.Hash), info.Type, info.Size)
} }
} else if response, ok := message.(*protocol.MessageResponse); ok { } else if response, ok := message.(*protocol.MessageResponse); ok {
output += fmt.Sprintf("Fields:\n Protocol supported %d\n", response.Protocol) text += fmt.Sprintf("Fields:\n Protocol supported %d\n", response.Protocol)
output += fmt.Sprintf(" Feature bits %d\n", response.Features) text += fmt.Sprintf(" Feature bits %d\n", response.Features)
output += fmt.Sprintf(" Action bits %d\n", response.Actions) text += fmt.Sprintf(" Action bits %d\n", response.Actions)
output += fmt.Sprintf(" Blockchain Height %d\n", response.BlockchainHeight) text += fmt.Sprintf(" Blockchain Height %d\n", response.BlockchainHeight)
output += fmt.Sprintf(" Blockchain Version %d\n", response.BlockchainVersion) text += fmt.Sprintf(" Blockchain Version %d\n", response.BlockchainVersion)
output += fmt.Sprintf(" Port Internal %d\n", response.PortInternal) text += fmt.Sprintf(" Port Internal %d\n", response.PortInternal)
output += fmt.Sprintf(" Port External %d\n", response.PortExternal) text += fmt.Sprintf(" Port External %d\n", response.PortExternal)
output += fmt.Sprintf(" User Agent %s\n", response.UserAgent) text += fmt.Sprintf(" User Agent %s\n", response.UserAgent)
for _, hash := range response.Hash2Peers { for _, hash := range response.Hash2Peers {
isLast := "" isLast := ""
if hash.IsLast { if hash.IsLast {
isLast = " [last result in sequence]" isLast = " [last result in sequence]"
} }
output += fmt.Sprintf(" - Peers known for the hash %s%s\n", hex.EncodeToString(hash.ID.Hash), isLast) text += fmt.Sprintf(" - Peers known for the hash %s%s\n", hex.EncodeToString(hash.ID.Hash), isLast)
for n := range hash.Closest { for n := range hash.Closest {
output += fmt.Sprintf(" Close peer:\n%s\n", outputPeerRecord(&hash.Closest[n])) text += fmt.Sprintf(" Close peer:\n%s\n", outputPeerRecord(&hash.Closest[n]))
} }
for n := range hash.Storing { for n := range hash.Storing {
output += fmt.Sprintf(" Peer stores:\n%s\n", outputPeerRecord(&hash.Storing[n])) text += fmt.Sprintf(" Peer stores:\n%s\n", outputPeerRecord(&hash.Storing[n]))
} }
} }
for _, find := range response.FilesEmbed { for _, find := range response.FilesEmbed {
output += fmt.Sprintf(" - File embedded %s (%d bytes)\n", hex.EncodeToString(find.ID.Hash), len(find.Data)) text += fmt.Sprintf(" - File embedded %s (%d bytes)\n", hex.EncodeToString(find.ID.Hash), len(find.Data))
} }
for _, hash := range response.HashesNotFound { for _, hash := range response.HashesNotFound {
output += fmt.Sprintf(" - Hash not found %s\n", hex.EncodeToString(hash)) text += fmt.Sprintf(" - Hash not found %s\n", hex.EncodeToString(hash))
} }
} else if traverse, ok := message.(*protocol.MessageTraverse); ok { } else if traverse, ok := message.(*protocol.MessageTraverse); ok {
output += fmt.Sprintf("Fields:\n Target Peer %s\n", hex.EncodeToString(traverse.TargetPeer.SerializeCompressed())) text += 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())) text += 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())) text += fmt.Sprintf(" Signer Public Key %s\n", hex.EncodeToString(traverse.SignerPublicKey.SerializeCompressed()))
output += fmt.Sprintf(" Expires %s\n", traverse.Expires.String()) text += fmt.Sprintf(" Expires %s\n", traverse.Expires.String())
output += fmt.Sprintf(" IPv4 %s\n", traverse.IPv4.String()) text += fmt.Sprintf(" IPv4 %s\n", traverse.IPv4.String())
output += fmt.Sprintf(" Port IPv4 %d\n", traverse.PortIPv4) text += fmt.Sprintf(" Port IPv4 %d\n", traverse.PortIPv4)
output += fmt.Sprintf(" Port IPv4 Reported External %d\n", traverse.PortIPv4ReportedExternal) text += fmt.Sprintf(" Port IPv4 Reported External %d\n", traverse.PortIPv4ReportedExternal)
output += fmt.Sprintf(" IPv6 %s\n", traverse.IPv6.String()) text += fmt.Sprintf(" IPv6 %s\n", traverse.IPv6.String())
output += fmt.Sprintf(" Port IPv6 %d\n", traverse.PortIPv6) text += fmt.Sprintf(" Port IPv6 %d\n", traverse.PortIPv6)
output += fmt.Sprintf(" Port IPv6 Reported External %d\n", traverse.PortIPv6ReportedExternal) text += fmt.Sprintf(" Port IPv6 Reported External %d\n", traverse.PortIPv6ReportedExternal)
} }
output += "--------\n" text += "--------\n"
fmt.Printf("%s", output) output.Write([]byte(text))
} }
func outputPeerRecord(record *protocol.PeerRecord) (output string) { func outputPeerRecord(record *protocol.PeerRecord) (output string) {
@@ -286,7 +296,8 @@ func outputPeerRecord(record *protocol.PeerRecord) (output string) {
} }
func outputOutgoingMessage(peer *core.PeerInfo, packet *protocol.PacketRaw) { func outputOutgoingMessage(peer *core.PeerInfo, packet *protocol.PacketRaw) {
if !hashIsMonitored(peer.NodeID) { monitored, output := hashIsMonitored(peer.NodeID)
if !monitored {
// TODO: For Announcement/Response also check data, Traverse the final target // TODO: For Announcement/Response also check data, Traverse the final target
return return
} }
@@ -310,14 +321,14 @@ func outputOutgoingMessage(peer *core.PeerInfo, packet *protocol.PacketRaw) {
commandA = "Chat" commandA = "Chat"
} }
output := fmt.Sprintf("-------- Node %s Outgoing %s --------\n", hex.EncodeToString(peer.NodeID), commandA) text := 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())) text += fmt.Sprintf("Receiver Peer ID: %s\n", hex.EncodeToString(peer.PublicKey.SerializeCompressed()))
// TODO: Decoding of payload data (done by caller of this function) // TODO: Decoding of payload data (done by caller of this function)
output += "--------\n" text += "--------\n"
fmt.Printf("%s", output) output.Write([]byte(text))
} }
func filterMessageOutAnnouncement(receiverPublicKey *btcec.PublicKey, peer *core.PeerInfo, packet *protocol.PacketRaw, findSelf bool, findPeer []protocol.KeyHash, findValue []protocol.KeyHash, files []protocol.InfoStore) { func filterMessageOutAnnouncement(receiverPublicKey *btcec.PublicKey, peer *core.PeerInfo, packet *protocol.PacketRaw, findSelf bool, findPeer []protocol.KeyHash, findValue []protocol.KeyHash, files []protocol.InfoStore) {

24
Command Line.go
View File

@@ -58,7 +58,7 @@ func userCommands(backend *core.Backend, input io.Reader, output io.Writer, term
defer func() { // unmonitor hashes in case of terminate signal defer func() { // unmonitor hashes in case of terminate signal
for hash := range monitoredHashes { for hash := range monitoredHashes {
hashMonitorControl([]byte(hash), 1) hashMonitorControl([]byte(hash), 1, nil)
} }
}() }()
@@ -324,9 +324,15 @@ func userCommands(backend *core.Backend, input io.Reader, output io.Writer, term
debugCmdConnect(backend, nodeID, output) debugCmdConnect(backend, nodeID, output)
case "debug watch searches": case "debug watch searches":
fmt.Fprintf(output, "Enable (1) or disable (0) watching of all outgoing DHT searches? (current setting: %t)\n", enableMonitorAll) fmt.Fprintf(output, "Enable (1) or disable (0) watching of all outgoing DHT searches?\n")
if number, valid, terminate := getUserOptionInt(reader, terminateSignal); valid && number >= 0 && number <= 1 { if number, valid, terminate := getUserOptionInt(reader, terminateSignal); valid && number >= 0 && number <= 1 {
enableMonitorAll = number == 1 if number == 1 {
hashMonitorControl([]byte(keyMonitorAllSearches), 0, output)
monitoredHashes[keyMonitorAllSearches] = struct{}{}
} else {
hashMonitorControl([]byte(keyMonitorAllSearches), 1, output)
delete(monitoredHashes, keyMonitorAllSearches)
}
} else if terminate { } else if terminate {
return return
} else { } else {
@@ -334,9 +340,15 @@ func userCommands(backend *core.Backend, input io.Reader, output io.Writer, term
} }
case "debug watch incoming": case "debug watch incoming":
fmt.Fprintf(output, "Enable (1) or disable (0) watching of all incoming information requests? (current setting: %t)\n", enableWatchIncomingAll) fmt.Fprintf(output, "Enable (1) or disable (0) watching of all incoming information requests?\n")
if number, valid, terminate := getUserOptionInt(reader, terminateSignal); valid && number >= 0 && number <= 1 { if number, valid, terminate := getUserOptionInt(reader, terminateSignal); valid && number >= 0 && number <= 1 {
enableWatchIncomingAll = number == 1 if number == 1 {
hashMonitorControl([]byte(keyMonitorAllRequests), 0, output)
monitoredHashes[keyMonitorAllRequests] = struct{}{}
} else {
hashMonitorControl([]byte(keyMonitorAllRequests), 1, output)
delete(monitoredHashes, keyMonitorAllRequests)
}
} else if terminate { } else if terminate {
return return
} else { } else {
@@ -366,7 +378,7 @@ func userCommands(backend *core.Backend, input io.Reader, output io.Writer, term
break break
} }
added := hashMonitorControl(hash, 2) added := hashMonitorControl(hash, 2, output)
if added { if added {
monitoredHashes[string(hash)] = struct{}{} monitoredHashes[string(hash)] = struct{}{}
fmt.Fprintf(output, "The hash was added to the monitoring list.\n") fmt.Fprintf(output, "The hash was added to the monitoring list.\n")

2
Main.go
View File

@@ -46,8 +46,6 @@ func init() {
} }
os.Exit(status) os.Exit(status)
} }
monitorKeys = make(map[string]struct{})
} }
func main() { func main() {