stratux/main/network.go

332 wiersze
9.2 KiB
Go

package main
import (
"golang.org/x/net/icmp"
"golang.org/x/net/ipv4"
"io/ioutil"
"log"
"math/rand"
"net"
"os"
"strconv"
"strings"
"sync"
"time"
)
type networkMessage struct {
msg []byte
msgType uint8
queueable bool
ts time.Time
}
type networkConnection struct {
Conn *net.UDPConn
Ip string
Port uint32
Capability uint8
messageQueue [][]byte // Device message queue.
/*
Sleep mode/throttle variables. "sleep mode" is actually now just a very reduced packet rate, since we don't know positively
when a client is ready to accept packets - we just assume so if we don't receive ICMP Unreachable packets in 5 secs.
*/
lastUnreachable time.Time // Last time the device sent an ICMP Unreachable packet.
nextMessageTime time.Time // The next time that the device is "able" to receive a message.
numOverflows uint32 // Number of times the queue has overflowed - for calculating the amount to chop off from the queue.
}
var messageQueue chan networkMessage
var outSockets map[string]networkConnection
var dhcpLeases map[string]string
var netMutex *sync.Mutex
var totalNetworkMessagesSent uint32
var pingResponse map[string]time.Time // Last time an IP responded to an "echo" response.
const (
NETWORK_GDL90_STANDARD = 1
NETWORK_AHRS_FFSIM = 2
NETWORK_AHRS_GDL90 = 4
dhcp_lease_file = "/var/lib/dhcp/dhcpd.leases"
)
// Read the "dhcpd.leases" file and parse out IP/hostname.
func getDHCPLeases() (map[string]string, error) {
dat, err := ioutil.ReadFile(dhcp_lease_file)
ret := make(map[string]string)
if err != nil {
return ret, err
}
lines := strings.Split(string(dat), "\n")
open_block := false
block_ip := ""
for _, line := range lines {
spaced := strings.Split(line, " ")
if len(spaced) > 2 && spaced[0] == "lease" {
open_block = true
block_ip = spaced[1]
} else if open_block && len(spaced) >= 4 && spaced[2] == "client-hostname" {
hostname := strings.TrimRight(strings.TrimLeft(strings.Join(spaced[3:], " "), "\""), "\";")
ret[block_ip] = hostname
open_block = false
} else if open_block && strings.HasPrefix(spaced[0], "}") { // No hostname.
open_block = false
ret[block_ip] = ""
}
}
return ret, nil
}
func isSleeping(k string) bool {
ipAndPort := strings.Split(k, ":")
lastPing, ok := pingResponse[ipAndPort[0]]
// No ping response. Assume disconnected/sleeping device.
if !ok || time.Since(lastPing) > (10*time.Second) {
return true
}
if time.Since(outSockets[k].lastUnreachable) < (5 * time.Second) {
return true
}
return false
}
// Throttle mode for testing port open and giving some start-up time to the app.
// Throttling is 0.1% data rate for first 15 seconds.
func isThrottled(k string) bool {
return (rand.Int()%1000 != 0) && time.Since(outSockets[k].lastUnreachable) < (15*time.Second)
}
func sendToAllConnectedClients(msg networkMessage) {
netMutex.Lock()
defer netMutex.Unlock()
for k, netconn := range outSockets {
// Check if this port is able to accept the type of message we're sending.
if (netconn.Capability & msg.msgType) == 0 {
continue
}
// Send non-queueable messages immediately, or discard if the client is in sleep mode.
if !msg.queueable {
if !isSleeping(k) {
netconn.Conn.Write(msg.msg) // Write immediately.
totalNetworkMessagesSent++
}
} else {
if !isSleeping(k) {
netconn.numOverflows = 0 // Reset the overflow counter whenever the client is not sleeping so that we're not penalizing future sleepmodes.
}
// Queue the message if the message is "queueable".
if len(netconn.messageQueue) >= maxUserMsgQueueSize { // Too many messages queued? Drop the oldest.
log.Printf("%s:%d - message queue overflow.\n", netconn.Ip, netconn.Port)
netconn.numOverflows++
s := 2 * netconn.numOverflows // Double the amount we chop off on each overflow.
if int(s) >= len(netconn.messageQueue) {
netconn.messageQueue = make([][]byte, 0)
} else {
netconn.messageQueue = netconn.messageQueue[s:]
}
}
netconn.messageQueue = append(netconn.messageQueue, msg.msg)
outSockets[k] = netconn
}
}
}
// Just returns the number of DHCP leases for now.
func getNetworkStats() uint {
ret := uint(len(dhcpLeases))
globalStatus.Connected_Users = ret
return ret
}
// See who has a DHCP lease and make a UDP connection to each of them.
func refreshConnectedClients() {
netMutex.Lock()
defer netMutex.Unlock()
validConnections := make(map[string]bool)
t, err := getDHCPLeases()
if err != nil {
log.Printf("getDHCPLeases(): %s\n", err.Error())
return
}
dhcpLeases = t
// Client connected that wasn't before.
for ip, hostname := range dhcpLeases {
for _, networkOutput := range globalSettings.NetworkOutputs {
ipAndPort := ip + ":" + strconv.Itoa(int(networkOutput.Port))
if _, ok := outSockets[ipAndPort]; !ok {
log.Printf("client connected: %s:%d (%s).\n", ip, networkOutput.Port, hostname)
addr, err := net.ResolveUDPAddr("udp", ipAndPort)
if err != nil {
log.Printf("ResolveUDPAddr(%s): %s\n", ipAndPort, err.Error())
continue
}
outConn, err := net.DialUDP("udp", nil, addr)
if err != nil {
log.Printf("DialUDP(%s): %s\n", ipAndPort, err.Error())
continue
}
newq := make([][]byte, 0)
outSockets[ipAndPort] = networkConnection{Conn: outConn, Ip: ip, Port: networkOutput.Port, Capability: networkOutput.Capability, messageQueue: newq}
}
validConnections[ipAndPort] = true
}
}
// Client that was connected before that isn't.
for ipAndPort, conn := range outSockets {
if _, ok := validConnections[ipAndPort]; !ok {
log.Printf("removed connection %s.\n", ipAndPort)
conn.Conn.Close()
delete(outSockets, ipAndPort)
}
}
}
func messageQueueSender() {
secondTimer := time.NewTicker(5 * time.Second)
queueTimer := time.NewTicker(400 * time.Microsecond) // 2500 msg/sec
for {
select {
case msg := <-messageQueue:
sendToAllConnectedClients(msg)
case <-queueTimer.C:
netMutex.Lock()
for k, netconn := range outSockets {
if len(netconn.messageQueue) > 0 && !isSleeping(k) && !isThrottled(k) {
tmpConn := netconn
tmpConn.Conn.Write(tmpConn.messageQueue[0])
totalNetworkMessagesSent++
tmpConn.messageQueue = tmpConn.messageQueue[1:]
outSockets[k] = tmpConn
}
}
netMutex.Unlock()
case <-secondTimer.C:
getNetworkStats()
}
}
}
func sendMsg(msg []byte, msgType uint8, queueable bool) {
messageQueue <- networkMessage{msg: msg, msgType: msgType, queueable: queueable, ts: time.Now()}
}
func sendGDL90(msg []byte, queueable bool) {
sendMsg(msg, NETWORK_GDL90_STANDARD, queueable)
}
func monitorDHCPLeases() {
//TODO: inotify or dhcp event hook.
timer := time.NewTicker(30 * time.Second)
for {
select {
case <-timer.C:
refreshConnectedClients()
}
}
}
func icmpEchoSender(c *icmp.PacketConn) {
timer := time.NewTicker(5 * time.Second)
for {
<-timer.C
// Collect IPs.
ips := make(map[string]bool)
for k, _ := range outSockets {
ipAndPort := strings.Split(k, ":")
ips[ipAndPort[0]] = true
}
// Send to all IPs.
for ip, _ := range ips {
wm := icmp.Message{
Type: ipv4.ICMPTypeEcho, Code: 0,
Body: &icmp.Echo{
ID: os.Getpid() & 0xffff, Seq: 1,
Data: []byte("STRATUX"),
},
}
wb, err := wm.Marshal(nil)
if err != nil {
log.Printf("couldn't send ICMP Echo: %s\n", err.Error())
continue
}
if _, err := c.WriteTo(wb, &net.IPAddr{IP: net.ParseIP(ip)}); err != nil {
log.Printf("couldn't send ICMP Echo: %s\n", err.Error())
continue
}
totalNetworkMessagesSent++
}
}
}
// Monitor clients going in/out of sleep mode via ICMP unreachable packets.
func sleepMonitor() {
c, err := icmp.ListenPacket("ip4:icmp", "0.0.0.0")
if err != nil {
log.Printf("error listening for udp - sending data to all ports for all connected clients. err: %s", err)
return
}
go icmpEchoSender(c)
defer c.Close()
for {
buf := make([]byte, 1500)
n, peer, err := c.ReadFrom(buf)
if err != nil {
log.Printf("%s\n", err.Error())
continue
}
msg, err := icmp.ParseMessage(1, buf[:n])
if err != nil {
continue
}
ip := peer.String()
// Look for echo replies, mark it as received.
if msg.Type == ipv4.ICMPTypeEchoReply {
pingResponse[ip] = time.Now()
continue // No further processing needed.
}
// Only deal with ICMP Unreachable packets (since that's what iOS and Android seem to be sending whenever the apps are not available).
if msg.Type != ipv4.ICMPTypeDestinationUnreachable {
continue
}
// Packet parsing.
mb, err := msg.Body.Marshal(1)
if err != nil {
continue
}
if len(mb) < 28 {
continue
}
// The unreachable port.
port := (uint16(mb[26]) << 8) | uint16(mb[27])
ipAndPort := ip + ":" + strconv.Itoa(int(port))
netMutex.Lock()
p, ok := outSockets[ipAndPort]
if !ok {
// Can't do anything, the client isn't even technically connected.
netMutex.Unlock()
continue
}
p.lastUnreachable = time.Now()
outSockets[ipAndPort] = p
netMutex.Unlock()
}
}
func initNetwork() {
messageQueue = make(chan networkMessage, 1024) // Buffered channel, 1024 messages.
outSockets = make(map[string]networkConnection)
pingResponse = make(map[string]time.Time)
netMutex = &sync.Mutex{}
refreshConnectedClients()
go monitorDHCPLeases()
go messageQueueSender()
go sleepMonitor()
}