Fix lint errors

build_scripts/generate_macOS_launchctl_service_files.py

@@ -20,9 +20,9 @@
 LAUNCH_DAEMONS_DIR = "/Library/LaunchDaemons"
 DEFAULT_SERVICE_NAME = "com.thinkst.opencanary"
 CONFIG_FILE_BASENAME = "opencanary.conf"
-DEFAULT_CONFIG_DIR = importlib.resources.files(OPENCANARY).joinpath('data')
-USER_CONFIG_FILE = DEFAULT_CONFIG_DIR.joinpath('settings.json')
-DEFAULT_CONFIG_FILE = DEFAULT_CONFIG_DIR.joinpath('.opencanary.conf')
+DEFAULT_CONFIG_DIR = importlib.resources.files(OPENCANARY).joinpath("data")
+USER_CONFIG_FILE = DEFAULT_CONFIG_DIR.joinpath("settings.json")
+DEFAULT_CONFIG_FILE = DEFAULT_CONFIG_DIR.joinpath(".opencanary.conf")
 
 # opencanary dirs
 OPENCANARY_DIR = realpath(join(dirname(__file__), pardir))
@@ -52,13 +52,17 @@
     opencanary_config_file = USER_CONFIG_FILE
 else:
     if not DEFAULT_CONFIG_FILE.exists():
-        print(f"Neither a user 'settings.json' nor a default '.opencanary.conf' found in '{DEFAULT_CONFIG_DIR}'!")
+        print(
+            f"Neither a user 'settings.json' nor a default '.opencanary.conf' found in '{DEFAULT_CONFIG_DIR}'!"
+        )
         print("Exiting...")
         sys.exit()
 
     opencanary_config_file = DEFAULT_CONFIG_FILE
-    print(f"Using default config file......")
-    print(f"(Create a file at '{USER_CONFIG_FILE}' for individual settings beyond the command line arguments)")
+    print("Using default config file......")
+    print(
+        f"(Create a file at '{USER_CONFIG_FILE}' for individual settings beyond the command line arguments)"
+    )
 
 print(f"\nUsing base configuration file: '{opencanary_config_file}'")
 
@@ -210,5 +214,7 @@
 print(f"      Bootstrap script: ./{path.relpath(install_service_script)}")
 print(f"        Bootout script: ./{path.relpath(uninstall_service_script)}\n")
 print(f"                Config: ./{path.relpath(config_output_file)}")
-print(f"      Enabled canaries: {', '.join(args.canaries) if args.canaries else 'Nothing enabled!'}\n")
+print(
+    f"      Enabled canaries: {', '.join(args.canaries) if args.canaries else 'Nothing enabled!'}\n"
+)
 print(f"To install as a system service run:\n    'sudo {install_service_script}'\n")

opencanary/modules/llmnr.py

@@ -8,17 +8,24 @@
 from scapy.layers.llmnr import LLMNRQuery
 import random
 
+
 class LLMNR(DatagramProtocol):
     def startQueryLoop(self):
-            self.sendLLMNRQuery()
-            next_interval = self.factory.query_interval + random.uniform(-self.factory.query_splay, self.factory.query_splay)
-            reactor.callLater(next_interval, self.startQueryLoop)
+        self.sendLLMNRQuery()
+        next_interval = self.factory.query_interval + random.uniform(
+            -self.factory.query_splay, self.factory.query_splay
+        )
+        reactor.callLater(next_interval, self.startQueryLoop)
 
     def sendLLMNRQuery(self):
         # Craft an LLMNR query packet
         target_ip = "224.0.0.252"  # LLMNR multicast IP address
-        target_port = 5355         # LLMNR port
-        llmnr_packet = IP(dst=target_ip, ttl=1) / UDP(dport=target_port) / LLMNRQuery(qd=DNSQR(qname=self.factory.query_hostname))
+        target_port = 5355  # LLMNR port
+        llmnr_packet = (
+            IP(dst=target_ip, ttl=1)
+            / UDP(dport=target_port)
+            / LLMNRQuery(qd=DNSQR(qname=self.factory.query_hostname))
+        )
 
         # Send the packet, set verbose to False to suppress "Sent 1 packet" std-out
         send(llmnr_packet, verbose=False)
@@ -27,32 +34,46 @@ def datagramReceived(self, data, host_and_port):
         try:
             llmnr_response = DNS(data)
             # Decode bytes to string and remove trailing dot if present
-            received_hostname = llmnr_response.qd.qname.decode('utf-8').rstrip('.')
+            received_hostname = llmnr_response.qd.qname.decode("utf-8").rstrip(".")
             # If the received hostname matches the canary hostname, it's suspicous - log it
             if received_hostname == self.factory.query_hostname:
                 source_ip = host_and_port[0]
-                logdata = {"query_hostname": self.factory.query_hostname, "response": llmnr_response.summary()}
-                self.transport.getPeer = lambda: IPv4Address("UDP", source_ip, host_and_port[1])
-                self.factory.log(logdata=logdata, transport=self.transport, logtype=self.factory.logtype_query_response)
+                logdata = {
+                    "query_hostname": self.factory.query_hostname,
+                    "response": llmnr_response.summary(),
+                }
+                self.transport.getPeer = lambda: IPv4Address(
+                    "UDP", source_ip, host_and_port[1]
+                )
+                self.factory.log(
+                    logdata=logdata,
+                    transport=self.transport,
+                    logtype=self.factory.logtype_query_response,
+                )
 
         except Exception as e:
-                error_message = f"Error processing LLMNR response: {e}"
-                self.factory.log(error_message, level='error')
+            error_message = f"Error processing LLMNR response: {e}"
+            self.factory.log(error_message, level="error")
+
 
 class CanaryLLMNR(CanaryService):
-    NAME = 'llmnr'
+    NAME = "llmnr"
 
     def __init__(self, config=None, logger=None):
         super(CanaryLLMNR, self).__init__(config=config, logger=logger)
         self.logtype_query_response = logger.LOG_LLMNR_QUERY_RESPONSE
-        self.query_hostname = config.getVal('llmnr.hostname', default='DC03')
-        self.port = int(config.getVal('llmnr.port', default=5355))
-        self.query_interval = int(config.getVal('llmnr.query_interval', default=60))  # Interval in seconds
-        self.query_splay = int(config.getVal('llmnr.query_splay', default=5))  # Default splay in seconds
-        self.listen_addr = config.getVal('device.listen_addr', default='')
+        self.query_hostname = config.getVal("llmnr.hostname", default="DC03")
+        self.port = int(config.getVal("llmnr.port", default=5355))
+        self.query_interval = int(
+            config.getVal("llmnr.query_interval", default=60)
+        )  # Interval in seconds
+        self.query_splay = int(
+            config.getVal("llmnr.query_splay", default=5)
+        )  # Default splay in seconds
+        self.listen_addr = config.getVal("device.listen_addr", default="")
 
     def getService(self):
         f = LLMNR()
         f.factory = self
         reactor.callWhenRunning(f.startQueryLoop)
-        return internet.UDPServer(self.port, f, interface=self.listen_addr)
+        return internet.UDPServer(self.port, f, interface=self.listen_addr)