{

staleDuration = staleSeconds * 1000 * 1000 * 10; // in ns intervals

checkDuration = checkSeconds * 1000 * 1000 * 10; // in ns intervals

GetSystemTimeAsLargeInteger(&timeLastChecked);

{

LARGE_INTEGER curTime;

// periodically remove stale entries

GetSystemTimeAsLargeInteger(&curTime);

if (curTime.QuadPart - timeLastChecked.QuadPart > checkDuration) {

auto it = connectTrackerTimes.cbegin();

while (it != connectTrackerTimes.cend()) {

if (curTime.QuadPart - it->first > staleDuration) {

connectTracker.erase(it->second);

it = connectTrackerTimes.erase(it);

} else {

break; // as this is an ordered map, we have exhausted all stale items

}

}

auto it2 = acceptTrackerTimes.cbegin();

while (it2 != acceptTrackerTimes.cend()) {

if (curTime.QuadPart - it2->first > staleDuration) {

acceptTracker.erase(*(it2->second));

it2 = acceptTrackerTimes.erase(it2);

} else {

break; // as this is an ordered map, we have exhausted all stale items

}

}

}

GetSystemTimeAsLargeInteger(&timeLastChecked);

{

if (eventHdr == NULL) {

fprintf(stderr, "NetworkTracker::SeenConnect invalid params\n");

return 0;

}

pid_t pid = 0;

PSYSMON_LINUX_NETWORK_EVENT event = (PSYSMON_LINUX_NETWORK_EVENT)&eventHdr->m_EventBody;

if (event->m_NewState == TCP_CLOSE) {

// remove state

auto it = connectTracker.find(event->m_SockId);

if (it != connectTracker.end()) {

connectTrackerTimes.erase(it->second.second);

connectTracker.erase(it);

}

pid = 0;

} else {

// attempt to insert - bool indicates success/failure

auto ret = connectTracker.emplace(event->m_SockId, std::pair<pid_t, LONGLONG>(event->m_ProcessId, event->m_EventTime.QuadPart));

if (ret.second) {

// successful insertion - didn't already exist

connectTrackerTimes.emplace(event->m_EventTime.QuadPart, event->m_SockId);

pid = 0;

} else {

// already exists

pid = ret.first->second.first;

if (event->m_NewState == TCP_ESTABLISHED) {

// connection is established, so remove it

connectTracker.erase(event->m_SockId);

connectTrackerTimes.erase(event->m_EventTime.QuadPart);

}

}

}

PurgeStale();

return pid;

{

acceptTracker.erase(sourceAddrAndPort);

auto ret = acceptTracker.emplace(sourceAddrAndPort, std::pair<AddrAndPort, LONGLONG>(destAddrAndPort, eventTime));

acceptTrackerTimes.emplace(eventTime, &(ret.first->first));

PurgeStale();

{

if (destAddrAndPort == NULL) {

fprintf(stderr, "NetworkTracker::SeenAccept invalid params\n");

return false;

}

auto it = acceptTracker.find(sourceAddrAndPort);

if (it == acceptTracker.end()) {

return false;

}

acceptTrackerTimes.erase(it->second.second);

memcpy(destAddrAndPort, &it->second.first, sizeof(AddrAndPort));

acceptTracker.erase(it);

return true;

{

auto it = acceptTracker.find(sourceAddrAndPort);

if (it == acceptTracker.end()) {

return;

}

if (!(it->second.first == destAddrAndPort)) {

return;

}

acceptTrackerTimes.erase(it->second.second);

acceptTracker.erase(it);

{

LONGLONG lastTime = 0;

auto it = udpRecvTrackerPidFds.find(pid);

if (it == udpRecvTrackerPidFds.end())

// no existing pid entry

return 0;

auto it2 = it->second.find(fd);

if (it2 == it->second.end())

// no existing fd entry

return 0;

if (p == NULL || it2->second.second != *p) {

// packet doesn't match

lastTime = 0;

} else {

lastTime = it2->second.first;

}

if (erase) {

// remove entries

it->second.erase(it2);

if (it->second.empty()) {

udpRecvTrackerPidFds.erase(it);

}

}

return lastTime;

{

if (p == NULL) {

fprintf(stderr, "NetworkTracker::FindOrEraseUdpPidAddr invalid params\n");

return 0;

}

LONGLONG lastTime = 0;

auto it = udpSendTrackerPidAddrs.find(pid);

if (it == udpSendTrackerPidAddrs.end())

// no existing pid entry

return 0;

auto it2 = it->second.find(*p);

if (it2 == it->second.end())

// no existing address entry

return 0;

lastTime = it2->second;

if (erase) {

// remove entries

it->second.erase(it2);

if (it->second.empty()) {

udpSendTrackerPidAddrs.erase(it);

}

}

return lastTime;

{

if (p == NULL) {

fprintf(stderr, "NetworkTracker::UpdateUdpPidFd invalid params\n");

return 0;

}

LONGLONG lastTime = 0;

std::pair<LONGLONG, PacketAddresses> newEntry(ctime, *p);

auto it = udpRecvTrackerPidFds.find(pid);

if (it == udpRecvTrackerPidFds.end()) {

// no existing pid entry

udpRecvTrackerPidFds.emplace(pid, std::unordered_map<int, std::pair<LONGLONG, PacketAddresses>>({{fd, newEntry}}));

return 0;

}

auto it2 = it->second.find(fd);

if (it2 == it->second.end()) {

// no existing fd entry

it->second.emplace(fd, newEntry);

return 0;

}

// update entry and return previous time

if (it2->second.second == *p) {

lastTime = it2->second.first;

it2->second.first = ctime;

return lastTime;

}

// entry was for a previous socket

it->second.erase(it2);

it->second.emplace(fd, newEntry);

return 0;

{

if (p == NULL) {

fprintf(stderr, "NetworkTracker::UpdateUdpPidAddr invalid params\n");

return 0;

}

LONGLONG lastTime = 0;

auto it = udpSendTrackerPidAddrs.find(pid);

if (it == udpSendTrackerPidAddrs.end()) {

// no existing pid entry

udpSendTrackerPidAddrs.emplace(pid, std::unordered_map<PacketAddresses, LONGLONG>({{*p, ctime}}));

return 0;

}

auto it2 = it->second.find(*p);

if (it2 == it->second.end()) {

// no existing address entry

it->second.emplace(*p, ctime);

return 0;

}

// update entry and return previous time

lastTime = it2->second;

it2->second = ctime;

return lastTime;

{

LONGLONG lastTime = 0;

pid_t pid;

int fd;

// traverse all PID/FD entries chronologically

auto it = udpRecvTrackerTimes.cbegin();

while (it != udpRecvTrackerTimes.cend()) {

// if an entry is older than our reporting interval

if (curTime - it->first > UDP_REPORT_INTERVAL) {

pid = it->second >> 32;

fd = it->second & 0xFFFFFFFF;

// check if the kernel version is older than our reporting interval or not

if (telemetryMapLookupElem(mapFds[UDP_PIDFD_HASH], &it->second, &lastTime) < 0) {

FindOrEraseUdpPidFd(pid, fd, NULL, true);

} else {

if (curTime - lastTime > UDP_REPORT_INTERVAL) {

// remove the element from the kernel hash

telemetryMapDeleteElem(mapFds[UDP_PIDFD_HASH], &it->second);

// remove from the pid and fd tracker

FindOrEraseUdpPidFd(pid, fd, NULL, true);

} else {

// update our records with latest seen time

PacketAddresses p;

if (GetUdp(&p, pid, fd)) {

UpdateUdpPidFd(pid, fd, lastTime, &p);

udpRecvTrackerTimes.emplace(lastTime, it->second);

} else {

FindOrEraseUdpPidFd(pid, fd, NULL, true);

}

}

}

// remove existing time entry and iterate

it = udpRecvTrackerTimes.erase(it);

} else {

break; // as this is an ordered map, we have exhausted all stale items

}

}

// traverse all packetAddrs entries chronologically

auto it2 = udpSendTrackerTimes.cbegin();

while (it2 != udpSendTrackerTimes.cend()) {

// if an entry is older than our reporting interval

if (curTime - it2->first > UDP_REPORT_INTERVAL) {

PacketAddresses p(it2->second.IPv4, it2->second.srcAddr, it2->second.srcPort,

it2->second.dstAddr, it2->second.dstPort);

// check if the kernel version is older than our reporting interval or not

if (telemetryMapLookupElem(mapFds[UDP_ADDRS_HASH], &it2->second, &lastTime) < 0) {

FindOrEraseUdpPidAddr(pid, &p, true);

} else {

if (curTime - lastTime > UDP_REPORT_INTERVAL) {

// remove the element from the kernel hash

telemetryMapDeleteElem(mapFds[UDP_ADDRS_HASH], &it2->second);

// remove from the pid and addr tracker

FindOrEraseUdpPidAddr(pid, &p, true);

} else {

// update our records with latest seen time

UpdateUdpPidAddr(pid, lastTime, &p);

udpSendTrackerTimes.emplace(lastTime, it2->second);

}

}

// remove existing time entry and iterate

it2 = udpSendTrackerTimes.erase(it2);

} else {

break; // as this is an ordered map, we have exhausted all stale items

}

}

{

if (path == NULL) {

fprintf(stderr, "NetworkTracker::PathToInode invalid params\n");

return 0;

}

if (strncmp(path, INODE_SOCK_PRE1, strlen(INODE_SOCK_PRE1)) == 0) {

return atoi(path + strlen(INODE_SOCK_PRE1));

} else if (strncmp(path, INODE_SOCK_PRE2, strlen(INODE_SOCK_PRE2)) == 0) {

return atoi(path + strlen(INODE_SOCK_PRE2));

}

return 0;

{

if (p == NULL) {

fprintf(stderr, "NetworkTracker::InodeToAddr invalid params\n");

return false;

}

FILE *fp = NULL;

char *line = NULL;

size_t len = 0;

ssize_t readLen;

int numTokens = 0;

char lAddrText[33];

char rAddrText[33];

uint32_t sl, st, tr, uid, timeout;

uint64_t txQueue, rxQueue, tmWhen, retrnsmt, inode;

char file[PATH_MAX];

if (IPv4) {

snprintf(file, PATH_MAX, "%s", PROC_UDP_FNAME);

} else {

snprintf(file, PATH_MAX, "%s", PROC_UDP6_FNAME);

}

memset(p->localAddr, 0, sizeof(p->localAddr));

memset(p->remoteAddr, 0, sizeof(p->remoteAddr));

fp = fopen(file, "r");

if (fp == NULL)

return false;

readLen = getline(&line, &len, fp); // skip first line

while ((readLen = getline(&line, &len, fp)) >= 0) {

numTokens = sscanf(line, "%u: %32[0-9A-Fa-f]:%hX %32[0-9A-Fa-f]:%hX %X %lX:%lX %X:%lX %lX %u %u %lu %*s\n",

&sl, lAddrText, &p->localPort, rAddrText, &p->remotePort, &st, &txQueue, &rxQueue, &tr, &tmWhen,

&retrnsmt, &uid, &timeout, &inode);

if (numTokens < 14 || inode != inodeIn)

continue;

if (strlen(lAddrText) == 8) {

// IPv4

p->IPv4 = true;

*(uint32_t *)p->localAddr = strtoul(lAddrText, NULL, 16);

*(uint32_t *)p->remoteAddr = strtoul(rAddrText, NULL, 16);

free(line);

fclose(fp);

return true;

}

if (strlen(lAddrText) == 32) {

// IPv6

p->IPv4 = false;

sscanf(lAddrText, "%08X%08X%08X%08X", (uint32_t *)p->localAddr, (uint32_t *)(p->localAddr + 4),

(uint32_t *)(p->localAddr + 8), (uint32_t *)(p->localAddr + 12));

sscanf(rAddrText, "%08X%08X%08X%08X", (uint32_t *)p->remoteAddr, (uint32_t *)(p->remoteAddr + 4),

(uint32_t *)(p->remoteAddr + 8), (uint32_t *)(p->remoteAddr + 12));

free(line);

fclose(fp);

return true;

}

}

free(line);

fclose(fp);

return false;

{

if (p == NULL) {

fprintf(stderr, "NetworkTracker::GetUdp invalid params\n");

return false;

}

char fdPath[PATH_MAX];

char fdSocket[PATH_MAX];

uint64_t inode = 0;

snprintf(fdPath, sizeof(fdPath), "/proc/%d/fd/%d", pid, fd);

if (readlink(fdPath, fdSocket, sizeof(fdSocket)) == -1)

return false;

inode = PathToInode(fdSocket);

if (inode == 0)

return false;

if (InodeToAddr(p, true, inode))

return true;

if (InodeToAddr(p, false, inode))

return true;

return false;

{

if (p == NULL) {

fprintf(stderr, "NetworkTracker::SeenUdp invalid params\n");

return false;

}

LARGE_INTEGER curTime;

LONGLONG lastTime;

uint64_t pidFd = ((uint64_t)pid << 32) | fd;

GetSystemTimeAsLargeInteger(&curTime);

if (!GetUdp(p, pid, fd)) {

// failed to look up in /proc

return false;

}

// get the most recent sent time from the kernel

if (telemetryMapLookupElem(mapFds[UDP_PIDFD_HASH], &pidFd, &lastTime) < 0) {

// kernel has sent this entry but it's not in the hash. This means

// the hash is full. So rapidly empty a significant proportion of it

// to make space for additional observations.

auto it = udpRecvTrackerTimes.cbegin();

for (unsigned int i=0; i<UDP_HASH_SIZE * UDP_HASH_RPP && it != udpRecvTrackerTimes.cend(); i++) {

telemetryMapDeleteElem(mapFds[UDP_PIDFD_HASH], &it->second);

it++;

}

// insert the one we have just received

// find the last time we sent it

lastTime = FindOrEraseUdpPidFd(pid, fd, p, false);

if (lastTime > 0) {

// copy to kernel table

telemetryMapUpdateElem(mapFds[UDP_PIDFD_HASH], &pidFd, &lastTime, MAP_UPDATE_CREATE_OR_OVERWRITE);

} else {

// insert new element

telemetryMapUpdateElem(mapFds[UDP_PIDFD_HASH], &pidFd, &curTime.QuadPart, MAP_UPDATE_CREATE_OR_OVERWRITE);

UpdateUdpPidFd(pid, fd, curTime.QuadPart, p);

udpRecvTrackerTimes.emplace(curTime.QuadPart, pidFd);

}

PurgeUdp(curTime.QuadPart);

return (curTime.QuadPart - lastTime > UDP_REPORT_INTERVAL);

}

LONGLONG prevTime = FindOrEraseUdpPidFd(pid, fd, p, false);

if (prevTime > 0) {

if (curTime.QuadPart - prevTime > UDP_REPORT_INTERVAL) {

// not sent this entry for long enough - update records and send event

telemetryMapUpdateElem(mapFds[UDP_PIDFD_HASH], &pidFd, &curTime.QuadPart, MAP_UPDATE_CREATE_OR_OVERWRITE);

lastTime = UpdateUdpPidFd(pid, fd, curTime.QuadPart, p);

udpRecvTrackerTimes.erase(lastTime);

udpRecvTrackerTimes.emplace(curTime.QuadPart, pidFd);

PurgeUdp(curTime.QuadPart);

return true;

} else {

// our local entry is older than the kernel one - update it

telemetryMapUpdateElem(mapFds[UDP_PIDFD_HASH], &pidFd, &prevTime, MAP_UPDATE_CREATE_OR_OVERWRITE);

PurgeUdp(curTime.QuadPart);

return false;

}

} else {

// we don't have an existing entry

UpdateUdpPidFd(pid, fd, lastTime, p);

udpRecvTrackerTimes.emplace(lastTime, pidFd);

PurgeUdp(curTime.QuadPart);

return true;

}

{

if (p == NULL) {

fprintf(stderr, "NetworkTracker::SeenUdp invalid params\n");

return false;

}

LARGE_INTEGER curTime;

LONGLONG lastTime;

packetAddrs pa;

pa.IPv4 = p->IPv4;

pa.srcPort = p->localPort;

pa.dstPort = p->remotePort;

memcpy(pa.srcAddr, p->localAddr, sizeof(pa.srcAddr));

memcpy(pa.dstAddr, p->remoteAddr, sizeof(pa.dstAddr));

GetSystemTimeAsLargeInteger(&curTime);

// get the most recent sent time from the kernel

if (telemetryMapLookupElem(mapFds[UDP_ADDRS_HASH], &pa, &lastTime) < 0) {

// kernel has sent this entry but it's not in the hash. This means

// the hash is full. So rapidly empty a significant proportion of it

// to make space for additional observations.

auto it = udpSendTrackerTimes.cbegin();

for (unsigned int i=0; i<UDP_HASH_SIZE * UDP_HASH_RPP && it != udpSendTrackerTimes.cend(); i++) {

telemetryMapDeleteElem(mapFds[UDP_PIDFD_HASH], &it->second);

it++;

}

// insert the one we have just received

// find the last time we sent it

lastTime = FindOrEraseUdpPidAddr(pid, p, false);

if (lastTime > 0) {

// copy to kernel table

telemetryMapUpdateElem(mapFds[UDP_ADDRS_HASH], &pa, &lastTime, MAP_UPDATE_CREATE_OR_OVERWRITE);

} else {

// insert new element

telemetryMapUpdateElem(mapFds[UDP_ADDRS_HASH], &pa, &curTime.QuadPart, MAP_UPDATE_CREATE_OR_OVERWRITE);

UpdateUdpPidAddr(pid, curTime.QuadPart, p);

udpSendTrackerTimes.emplace(curTime.QuadPart, pa);

}

PurgeUdp(curTime.QuadPart);

return (curTime.QuadPart - lastTime > UDP_REPORT_INTERVAL);

}

LONGLONG prevTime = FindOrEraseUdpPidAddr(pid, p, false);

if (prevTime > 0) {

if (curTime.QuadPart - prevTime > UDP_REPORT_INTERVAL) {

// not sent this entry for long enough - update records and send event

telemetryMapUpdateElem(mapFds[UDP_ADDRS_HASH], &pa, &curTime.QuadPart, MAP_UPDATE_CREATE_OR_OVERWRITE);

lastTime = UpdateUdpPidAddr(pid, curTime.QuadPart, p);

udpSendTrackerTimes.erase(lastTime);

udpSendTrackerTimes.emplace(curTime.QuadPart, pa);

PurgeUdp(curTime.QuadPart);

return true;

} else {

// our local entry is older than the kernel one - update it

telemetryMapUpdateElem(mapFds[UDP_ADDRS_HASH], &pa, &prevTime, MAP_UPDATE_CREATE_OR_OVERWRITE);

PurgeUdp(curTime.QuadPart);

return false;

}

} else {

// we don't have an existing entry

UpdateUdpPidAddr(pid, lastTime, p);

udpSendTrackerTimes.emplace(lastTime, pa);

PurgeUdp(curTime.QuadPart);

return true;

}

{

auto it = udpRecvTrackerPidFds.find(pid);

if (it == udpRecvTrackerPidFds.end())

// no existing pid entry

return;

auto it2 = it->second.begin();

while (it2 != it->second.end()) {

LONGLONG lastTime = it2->second.first;

auto it3 = udpRecvTrackerTimes.find(lastTime);

if (it3 != udpRecvTrackerTimes.end()) {

telemetryMapDeleteElem(mapFds[UDP_PIDFD_HASH], &it3->second);

}

udpRecvTrackerTimes.erase(lastTime);

it2 = it->second.erase(it2);

}

udpRecvTrackerPidFds.erase(it);

auto it4 = udpSendTrackerPidAddrs.find(pid);

if (it4 == udpSendTrackerPidAddrs.end())

// no existing pid entry

return;

auto it5 = it4->second.begin();

while (it5 != it4->second.end()) {

LONGLONG lastTime = it5->second;

auto it6 = udpSendTrackerTimes.find(lastTime);

if (it6 != udpSendTrackerTimes.end()) {

telemetryMapDeleteElem(mapFds[UDP_ADDRS_HASH], &it6->second);

}

udpSendTrackerTimes.erase(lastTime);

it5 = it4->second.erase(it5);

}

{

if (n == NULL || eventHdr == NULL) {

fprintf(stderr, "NetworkTrackerSeenConnect invalid params\n");

return 0;

}

return n->SeenConnect(eventHdr);

unsigned short sourcePort, const BYTE *destAddr, unsigned short destPort, LONGLONG eventTime)

{

if (n == NULL || sourceAddr == NULL || destAddr == NULL) {

fprintf(stderr, "NetworkTrackerSeenFullAccept invalid params\n");

return;

}

n->SeenAccept(AddrAndPort(sourceAddr, IPv4, sourcePort), AddrAndPort(destAddr, IPv4, destPort), eventTime);

unsigned short sourcePort, BYTE *destAddr, unsigned short *destPort)

{

if (n == NULL || sourceAddr == NULL || destAddr == NULL) {

fprintf(stderr, "NetworkTrackerSeenAccept invalid params\n");

return false;

}

BYTE empty[16] = {0};

AddrAndPort d(empty, IPv4, 0);

if (n->SeenAccept(AddrAndPort(sourceAddr, IPv4, sourcePort), &d)) {

memcpy(destAddr, d.addr, sizeof(d.addr));

*destPort = d.port;

return true;

}

return false;

unsigned short sourcePort, const BYTE *destAddr, unsigned short destPort)

{

if (n == NULL || sourceAddr == NULL || destAddr == NULL) {

fprintf(stderr, "NetworkTrackerCloseAccept invalid params\n");

return;

}

n->CloseAccept(AddrAndPort(sourceAddr, IPv4, sourcePort), AddrAndPort(destAddr, IPv4, destPort));

unsigned short *sourcePort, BYTE *destAddr, unsigned short *destPort, pid_t pid, int fd)

{

if (n == NULL || sourceAddr == NULL || sourcePort == NULL || destAddr == NULL || destPort == NULL) {

fprintf(stderr, "NetworkTrackerSeenUdpRecv invalid params\n");

return false;

}

PacketAddresses p;

if (!n->SeenUdp(&p, pid, fd))

return false;

*IPv4 = p.IPv4;

*sourcePort = p.localPort;

*destPort = p.remotePort;

memcpy(sourceAddr, p.localAddr, sizeof(p.localAddr));

memcpy(destAddr, p.remoteAddr, sizeof(p.remoteAddr));

return true;

unsigned short sourcePort, const BYTE *destAddr, unsigned short destPort, pid_t pid)

{

if (n == NULL || sourceAddr == NULL || destAddr == NULL) {

fprintf(stderr, "NetworkTrackerSeenUdpSend invalid params\n");

return false;

}

PacketAddresses p(IPv4, sourceAddr, sourcePort, destAddr, destPort);

return n->SeenUdp(&p, pid);

{

if (n == NULL) {

fprintf(stderr, "NetworkTrackerUdpProgramTermination invalid params\n");

return;

}

n->UdpProgramTermination(pid);