Ability to block user agents

daemon.py

@@ -457,20 +457,34 @@ class PubServer(BaseHTTPRequestHandler):
         """Should a GET or POST be blocked based upon its user agent?
         """
         agentDomain = None
+        agentStr = None
         if self.headers.get('User-Agent'):
             agentStr = self.headers['User-Agent']
+            # is this a web crawler? If so the block it
             agentStrLower = agentStr.lower()
             if 'bot/' in agentStrLower or 'bot-' in agentStrLower:
                 print('Crawler: ' + agentStr)
                 return True
+            # get domain name from User-Agent
             agentDomain = userAgentDomain(agentStr, self.server.debug)
         else:
+            # no User-Agent header is present
             return True
 
-        if not agentDomain:
+        # is the User-Agent type blocked? eg. "Mastodon"
-            if self.server.userAgentDomainRequired:
+        if self.server.userAgentsBlocked:
+            blockedUA = False
+            for agentName in self.server.userAgentsBlocked:
+                if agentName in agentStr:
+                    blockedUA = True
+                    break
+            if blockedUA:
                 return True
+
+        if not agentDomain:
             return False
+
+        # is the User-Agent domain blocked
         blockedUA = False
         if not agentDomain.startswith(callingDomain):
             blockedUA = isBlockedDomain(self.server.baseDir, agentDomain)
@@ -14878,7 +14892,7 @@ def loadTokens(baseDir: str, tokensDict: {}, tokensLookup: {}) -> None:
         break
 
 
-def runDaemon(userAgentDomainRequired: bool,
+def runDaemon(userAgentsBlocked: [],
               logLoginFailures: bool,
               city: str,
               showNodeInfoAccounts: bool,
@@ -15005,9 +15019,8 @@ def runDaemon(userAgentDomainRequired: bool,
     httpd.keyShortcuts = {}
     loadAccessKeysForAccounts(baseDir, httpd.keyShortcuts, httpd.accessKeys)
 
-    # if set to True then the calling domain must be specified
+    # list of blocked user agent types within the User-Agent header
-    # within the User-Agent header
+    httpd.userAgentsBlocked = userAgentsBlocked
-    httpd.userAgentDomainRequired = userAgentDomainRequired
 
     httpd.unitTest = unitTest
     httpd.allowLocalNetworkAccess = allowLocalNetworkAccess

epicyon.py

@@ -104,6 +104,9 @@ def str2bool(v) -> bool:
 
 
 parser = argparse.ArgumentParser(description='ActivityPub Server')
+parser.add_argument('--userAgentBlocks', type=str,
+                    default=None,
+                    help='List of blocked user agents, separated by commas')
 parser.add_argument('-n', '--nickname', dest='nickname', type=str,
                     default=None,
                     help='Nickname of the account to use')
@@ -274,12 +277,6 @@ parser.add_argument("--repliesEnabled", "--commentsEnabled",
                     type=str2bool, nargs='?',
                     const=True, default=True,
                     help="Enable replies to a post")
-parser.add_argument("--userAgentDomainRequired",
-                    dest='userAgentDomainRequired',
-                    type=str2bool, nargs='?',
-                    const=True, default=False,
-                    help="Whether User-Agent header must " +
-                    "contain the calling domain")
 parser.add_argument("--showPublishAsIcon",
                     dest='showPublishAsIcon',
                     type=str2bool, nargs='?',
@@ -2522,10 +2519,17 @@ showNodeInfoVersion = \
 if showNodeInfoVersion is not None:
     args.showNodeInfoVersion = bool(showNodeInfoVersion)
 
-userAgentDomainRequired = \
+userAgentsBlocked = []
-    getConfigParam(baseDir, 'userAgentDomainRequired')
+if args.userAgentBlocks:
-if userAgentDomainRequired is not None:
+    userAgentsBlockedStr = args.userAgentBlocks
-    args.userAgentDomainRequired = bool(userAgentDomainRequired)
+    setConfigParam(baseDir, 'userAgentsBlocked', userAgentsBlockedStr)
+else:
+    userAgentsBlockedStr = \
+        getConfigParam(baseDir, 'userAgentsBlocked')
+if userAgentsBlockedStr:
+    agentBlocksList = userAgentsBlockedStr.split(',')
+    for agentBlockStr in agentBlocksList:
+        userAgentsBlocked.append(agentBlockStr.strip())
 
 city = \
     getConfigParam(baseDir, 'city')
@@ -2563,7 +2567,7 @@ if args.registration:
         print('New registrations closed')
 
 if __name__ == "__main__":
-    runDaemon(args.userAgentDomainRequired,
+    runDaemon(userAgentsBlocked,
               args.logLoginFailures,
               args.city,
               args.showNodeInfoAccounts,

tests.py

@@ -520,9 +520,9 @@ def createServerAlice(path: str, domain: str, port: int,
     showNodeInfoVersion = True
     city = 'London, England'
     logLoginFailures = False
-    userAgentDomainRequired = False
+    userAgentsBlocked = []
     print('Server running: Alice')
-    runDaemon(userAgentDomainRequired,
+    runDaemon(userAgentsBlocked,
               logLoginFailures, city,
               showNodeInfoAccounts,
               showNodeInfoVersion,
@@ -625,9 +625,9 @@ def createServerBob(path: str, domain: str, port: int,
     showNodeInfoVersion = True
     city = 'London, England'
     logLoginFailures = False
-    userAgentDomainRequired = False
+    userAgentsBlocked = []
     print('Server running: Bob')
-    runDaemon(userAgentDomainRequired,
+    runDaemon(userAgentsBlocked,
               logLoginFailures, city,
               showNodeInfoAccounts,
               showNodeInfoVersion,
@@ -685,9 +685,9 @@ def createServerEve(path: str, domain: str, port: int, federationList: [],
     showNodeInfoVersion = True
     city = 'London, England'
     logLoginFailures = False
-    userAgentDomainRequired = False
+    userAgentsBlocked = []
     print('Server running: Eve')
-    runDaemon(userAgentDomainRequired,
+    runDaemon(userAgentsBlocked,
               logLoginFailures, city,
               showNodeInfoAccounts,
               showNodeInfoVersion,