packetlogging fixes, file writing optimizations (#17)

pkg/storage/iface.go

@@ -1,6 +1,9 @@
 package storage
 
-import "io"
+import (
+	"io"
+	"io/fs"
+)
 
 type Iface interface {
 	GetPath() string
@@ -10,6 +13,7 @@ type Iface interface {
 	Remove(name string) error
 	Rename(oldName, newName string) error
 	AppendFile(name string, data []byte) error
+	EnsurePermissions(name string, mode fs.FileMode) error
 	ReadWriter
 	Seeker
 }
@@ -21,6 +25,7 @@ type ReadWriter interface {
 	ConfigPath(filename string) string
 	OpenFile(name string) (io.ReadCloser, error)
 	OpenFileForWriting(name string) (io.WriteCloser, error)
+	OpenFileForAppending(name string) (io.WriteCloser, error)
 }
 
 type Seeker interface {

pkg/storage/local/path.go

@@ -3,6 +3,7 @@ package localstorage
 import (
 	"errors"
 	"fmt"
+	"io/fs"
 	"os"
 	"os/user"
 	"path"
@@ -85,3 +86,7 @@ func (l *LocalStorage) Remove(name string) error {
 func (l *LocalStorage) Rename(oldName, newName string) error {
 	return os.Rename(path.Join(l.path, oldName), path.Join(l.path, newName))
 }
+
+func (l *LocalStorage) EnsurePermissions(name string, mode fs.FileMode) error {
+	return os.Chmod(path.Join(l.path, name), mode)
+}

pkg/storage/local/write.go

@@ -12,7 +12,7 @@ func (l *LocalStorage) WriteFile(name string, data []byte) error {
 }
 
 func (l *LocalStorage) AppendFile(name string, data []byte) error {
-	f, err := os.OpenFile(path.Join(l.path, name), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0600)
+	f, err := os.OpenFile(path.Join(l.path, name), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0660)
 	if err != nil {
 		return err
 	}
@@ -31,3 +31,11 @@ func (l *LocalStorage) OpenFileForWriting(name string) (io.WriteCloser, error) {
 	}
 	return file, nil
 }
+
+func (l *LocalStorage) OpenFileForAppending(name string) (io.WriteCloser, error) {
+	file, err := os.OpenFile(path.Join(l.path, name), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0660)
+	if err != nil {
+		return nil, err
+	}
+	return file, nil
+}

pkg/storage/memory/storage.go

@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"fmt"
 	"io"
+	"io/fs"
 	"os"
 	"path"
 	"strings"
@@ -144,3 +145,18 @@ func (m *MockMemoryStorage) OpenFileForWriting(name string) (io.WriteCloser, err
 	m.Data[name] = (*MockReadWriterData)(&[]byte{})
 	return m.Data[name], nil
 }
+func (m *MockMemoryStorage) OpenFileForAppending(name string) (io.WriteCloser, error) {
+	if m.Data == nil {
+		m.Data = make(map[string]*MockReadWriterData)
+	}
+	val, ok := m.Data[name]
+	if !ok {
+		m.Data[name] = (*MockReadWriterData)(&[]byte{})
+		return m.Data[name], nil
+	}
+	m.Data[name] = (*MockReadWriterData)(val)
+	return m.Data[name], nil
+}
+func (m *MockMemoryStorage) EnsurePermissions(name string, mode fs.FileMode) error {
+	return nil
+}

pkg/wireguard/cache.go

@@ -22,9 +22,13 @@ func UpdateClientCache(peerConfig PeerConfig, clientCache *ClientCache) error {
 	}
 
 	if !found {
+		clientID, _, err := getClientIDAndConfigID(peerConfig.ID)
+		if err != nil {
+			return fmt.Errorf("can't parse peer config ID (%s): %s", peerConfig.ID, err)
+		}
 		clientCache.Addresses = append(clientCache.Addresses, ClientCacheAddresses{
 			Address:  *peerConfigAddressParsed,
-			ClientID: peerConfig.ID,
+			ClientID: clientID,
 		})
 	}
 

pkg/wireguard/packetlogger.go

@@ -59,6 +59,11 @@ func RunPacketLogger(storage storage.Iface, clientCache *ClientCache, vpnConfig
 		logging.ErrorLog(fmt.Errorf("could not ensure ownership of stats path: %s. Stats disabled", err))
 		return
 	}
+	err = storage.EnsurePermissions(path.Join(VPN_STATS_DIR, VPN_PACKETLOGGER_DIR), 0770|os.ModeSetgid)
+	if err != nil {
+		logging.ErrorLog(fmt.Errorf("could not ensure permissions of stats path: %s. Stats disabled", err))
+		return
+	}
 
 	useSyscalls := false
 	if runtime.GOOS == "darwin" {
@@ -71,33 +76,40 @@ func RunPacketLogger(storage storage.Iface, clientCache *ClientCache, vpnConfig
 	}
 	defer handle.Close()
 	i := 0
+	openFiles := make(PacketLoggerOpenFiles)
 	for {
-		err := readPacket(storage, handle, clientCache)
+		err := readPacket(storage, handle, clientCache, openFiles)
 		if err != nil {
 			logging.DebugLog(fmt.Errorf("readPacket error: %s", err))
 		}
 		if !vpnConfig.EnablePacketLogs {
 			logging.InfoLog("disabling packetlogs")
+			for _, openFile := range openFiles {
+				openFile.Close()
+			}
 			return
 		}
 		if i%1000 == 0 {
 			if err := checkDiskSpace(); err != nil {
 				logging.ErrorLog(fmt.Errorf("disk space error: %s", err))
+				for _, openFile := range openFiles {
+					openFile.Close()
+				}
 				return
 			}
 			i = 0
 		}
 		i++
 	}
 }
-func readPacket(storage storage.Iface, handle *pcap.Handle, clientCache *ClientCache) error {
+func readPacket(storage storage.Iface, handle *pcap.Handle, clientCache *ClientCache, openFiles PacketLoggerOpenFiles) error {
 	data, _, err := handle.ReadPacketData()
 	if err != nil {
 		return fmt.Errorf("read packet error: %s", err)
 	}
-	return parsePacket(storage, data, clientCache)
+	return parsePacket(storage, data, clientCache, openFiles, time.Now())
 }
-func parsePacket(storage storage.Iface, data []byte, clientCache *ClientCache) error {
+func parsePacket(storage storage.Iface, data []byte, clientCache *ClientCache, openFiles PacketLoggerOpenFiles, now time.Time) error {
 	packet := gopacket.NewPacket(data, layers.IPProtocolIPv4, gopacket.DecodeOptions{Lazy: true, DecodeStreamsAsDatagrams: true})
 	var (
 		ip4   *layers.IPv4
@@ -129,19 +141,56 @@ func parsePacket(storage storage.Iface, data []byte, clientCache *ClientCache) e
 	if clientID == "" { // doesn't match a client ID
 		return nil
 	}
-	now := time.Now()
-	filename := path.Join(VPN_STATS_DIR, VPN_PACKETLOGGER_DIR, clientID+"-"+now.Format("2006-01-02")+".log")
+
+	// handle open files
+	logWriter, isFileOpen := openFiles[clientID+"-"+now.Format("2006-01-02")]
+	if !isFileOpen {
+		var err error
+		filename := path.Join(VPN_STATS_DIR, VPN_PACKETLOGGER_DIR, clientID+"-"+now.Format("2006-01-02")+".log")
+		// check if we need to close an older writer
+		for openFileKey, logWriterToClose := range openFiles {
+			filenameSplit := strings.Split(openFileKey, "-")
+			if len(filenameSplit) > 3 {
+				dateParsed, err := time.Parse("2006-01-02", strings.Join(filenameSplit[len(filenameSplit)-3:], "-"))
+				if err != nil {
+					logging.ErrorLog(fmt.Errorf("packetlogger: closing unknown open file %s (cannot parse date)", filename))
+					logWriterToClose.Close()
+					delete(openFiles, openFileKey)
+				} else {
+					if !dateutils.DateEqual(dateParsed, now) {
+						logWriterToClose.Close()
+						delete(openFiles, openFileKey)
+					}
+				}
+			} else {
+				logging.ErrorLog(fmt.Errorf("packetlogger: closing file without a date %s", filename))
+				logWriterToClose.Close()
+				delete(openFiles, openFileKey)
+			}
+		}
+		// open new file for appending
+		logWriter, err = storage.OpenFileForAppending(filename)
+		if err != nil {
+			return fmt.Errorf("could not open file for appending (%s): %s", clientID+"-"+now.Format("2006-01-02"), err)
+		}
+		err = storage.EnsurePermissions(filename, 0640)
+		if err != nil {
+			return fmt.Errorf("could not set permissions (%s): %s", clientID+"-"+now.Format("2006-01-02"), err)
+		}
+		openFiles[clientID+"-"+now.Format("2006-01-02")] = logWriter
+	}
+
 	if tcpLayer := packet.Layer(layers.LayerTypeTCP); tcpLayer != nil {
 		tcpPacket, _ := tcpLayer.(*layers.TCP)
 		if tcpPacket.SYN {
-			storage.AppendFile(filename, []byte(strings.Join([]string{
-				time.Now().Format(TIMESTAMP_FORMAT),
+			logWriter.Write([]byte(strings.Join([]string{
+				now.Format(TIMESTAMP_FORMAT),
 				"tcp",
 				srcIP.String(),
 				dstIP.String(),
 				strconv.FormatUint(uint64(tcpPacket.SrcPort), 10),
 				strconv.FormatUint(uint64(tcpPacket.DstPort), 10)},
-				",")+"\n",
+				",") + "\n",
 			))
 		}
 		switch tcpPacket.DstPort {
@@ -153,15 +202,15 @@ func parsePacket(storage storage.Iface, data []byte, clientCache *ClientCache) e
 					if err != nil {
 						fmt.Printf("debug: can't parse http packet: %s", err)
 					} else {
-						storage.AppendFile(filename, []byte(strings.Join([]string{
-							time.Now().Format(TIMESTAMP_FORMAT),
+						logWriter.Write([]byte(strings.Join([]string{
+							now.Format(TIMESTAMP_FORMAT),
 							"http",
 							srcIP.String(),
 							dstIP.String(),
 							strconv.FormatUint(uint64(tcpPacket.SrcPort), 10),
 							strconv.FormatUint(uint64(tcpPacket.DstPort), 10),
 							"http://" + req.Host + req.URL.RequestURI()},
-							",")+"\n",
+							",") + "\n",
 						))
 					}
 				}
@@ -170,15 +219,15 @@ func parsePacket(storage storage.Iface, data []byte, clientCache *ClientCache) e
 			if tls, ok := packet.Layer(layers.LayerTypeTLS).(*layers.TLS); ok {
 				for _, handshake := range tls.Handshake {
 					if sni := parseTLSExtensionSNI([]byte(handshake.ClientHello.Extensions)); sni != nil {
-						storage.AppendFile(filename, []byte(strings.Join([]string{
-							time.Now().Format(TIMESTAMP_FORMAT),
+						logWriter.Write([]byte(strings.Join([]string{
+							now.Format(TIMESTAMP_FORMAT),
 							"https",
 							srcIP.String(),
 							dstIP.String(),
 							strconv.FormatUint(uint64(tcpPacket.SrcPort), 10),
 							strconv.FormatUint(uint64(tcpPacket.DstPort), 10),
 							string(sni)},
-							",")+"\n",
+							",") + "\n",
 						))
 					}
 				}
@@ -205,15 +254,15 @@ func parsePacket(storage storage.Iface, data []byte, clientCache *ClientCache) e
 					}
 
 				}
-				storage.AppendFile(filename, []byte(strings.Join([]string{
-					time.Now().Format(TIMESTAMP_FORMAT),
+				logWriter.Write([]byte(strings.Join([]string{
+					now.Format(TIMESTAMP_FORMAT),
 					"udp",
 					srcIP.String(),
 					dstIP.String(),
 					strconv.FormatUint(uint64(udp.SrcPort), 10),
 					strconv.FormatUint(uint64(udp.DstPort), 10),
 					strings.Join(questions, "#")},
-					",")+"\n"))
+					",") + "\n"))
 			}
 		}
 	}