feat(commands:cloudflare|Network): add workaround to iplocalise for cloudflare

.idea/dictionaries/romain.xml

@@ -9,6 +9,7 @@
       <w>apres</w>
       <w>asctime</w>
       <w>commandstats</w>
+      <w>crimeflare</w>
       <w>ctype</w>
       <w>debian</w>
       <w>dnskey</w>

tuxbot/cogs/Network/functions/utils.py

@@ -1,10 +1,13 @@
 import socket
-from typing import Union, NoReturn
+from typing import Union, NoReturn, Optional
 
-import discord
+import asyncio
+import re
 import ipinfo
 import ipwhois
 import pydig
+import aiohttp
+
 from ipinfo.exceptions import RequestQuotaExceededError
 
 from ipwhois import Net
@@ -22,7 +25,7 @@ def _(x):
     return x
 
 
-async def get_ip(ip: str, inet: str = "", tmp: discord.Message = None) -> str:
+async def get_ip(ip: str, inet: str = "") -> str:
     _inet: Union[socket.AddressFamily, int] = 0  # pylint: disable=no-member
 
     if inet == "6":
@@ -33,9 +36,6 @@ async def get_ip(ip: str, inet: str = "", tmp: discord.Message = None) -> str:
     try:
         return socket.getaddrinfo(str(ip), None, _inet)[1][4][0]
     except socket.gaierror as e:
-        if tmp:
-            await tmp.delete()
-
         raise VersionNotFound(
             _(
                 "Unable to collect information on this in the given "
@@ -51,9 +51,7 @@ async def get_hostname(ip: str) -> str:
         return "N/A"
 
 
-async def get_ipwhois_result(
+async def get_ipwhois_result(ip_address: str) -> Union[NoReturn, dict]:
-    ip_address: str, tmp: discord.Message = None
-) -> Union[NoReturn, dict]:
     try:
         net = Net(ip_address)
         obj = IPASN(net)
@@ -61,9 +59,6 @@ async def get_ipwhois_result(
     except ipwhois.exceptions.ASNRegistryError:
         return {}
     except ipwhois.exceptions.IPDefinedError as e:
-        if tmp:
-            await tmp.delete()
-
         raise RFC18(
             _(
                 "IP address {ip_address} is already defined as Private-Use"
@@ -72,9 +67,7 @@ async def get_ipwhois_result(
         ) from e
 
 
-async def get_ipinfo_result(
+async def get_ipinfo_result(apikey: str, ip_address: str) -> dict:
-    apikey: str, ip_address: str
-) -> Union[NoReturn, dict]:
     try:
         handler = ipinfo.getHandlerAsync(
             apikey, request_options={"timeout": 7}
@@ -84,6 +77,25 @@ async def get_ipinfo_result(
         return {}
 
 
+async def get_crimeflare_result(
+    session: aiohttp.ClientSession, ip_address: str
+) -> Optional[str]:
+    try:
+        async with session.post(
+            "http://www.crimeflare.org:82/cgi-bin/cfsearch.cgi",
+            data=f"cfS={ip_address}",
+            timeout=aiohttp.ClientTimeout(total=15),
+        ) as s:
+            ip = re.search(r"(\d*\.\d*\.\d*\.\d*)", await s.text())
+
+            if ip:
+                return ip.group()
+    except (aiohttp.ClientError, asyncio.exceptions.TimeoutError):
+        pass
+
+    return None
+
+
 def merge_ipinfo_ipwhois(ipinfo_result: dict, ipwhois_result: dict) -> dict:
     output = {"belongs": "N/A", "rir": "N/A", "region": "N/A", "flag": "N/A"}
 

tuxbot/cogs/Network/network.py

@@ -6,6 +6,7 @@ from typing import Union, Optional
 import aiohttp
 import discord
 from aiohttp import ClientConnectorError
+from jishaku.models import copy_context_with
 from discord.ext import commands
 from ipinfo.exceptions import RequestQuotaExceededError
 from structured_config import ConfigFile
@@ -42,6 +43,7 @@ from .functions.utils import (
     get_pydig_result,
     check_query_type_or_raise,
     check_ip_version_or_raise,
+    get_crimeflare_result,
 )
 
 log = logging.getLogger("tuxbot.cogs.Network")
@@ -56,7 +58,7 @@ class Network(commands.Cog):
             NetworkConfig,
         ).config
 
-    async def cog_command_error(self, ctx, error):
+    async def cog_command_error(self, ctx: ContextPlus, error):
         if isinstance(
             error,
             (
@@ -70,6 +72,9 @@ class Network(commands.Cog):
         ):
             await ctx.send(_(str(error), ctx, self.bot.config))
 
+    async def cog_before_invoke(self, ctx: ContextPlus):
+        await ctx.trigger_typing()
+
     # =========================================================================
     # =========================================================================
 
@@ -82,18 +87,13 @@ class Network(commands.Cog):
     ):
         check_ip_version_or_raise(str(version))
 
-        tmp = await ctx.send(
+        ip_address = await get_ip(str(ip), str(version))
-            _("*Retrieving information...*", ctx, self.bot.config),
-            deletable=False,
-        )
-
-        ip_address = await get_ip(str(ip), str(version), tmp)
         ip_hostname = await get_hostname(ip_address)
 
         ipinfo_result = await get_ipinfo_result(
             self.__config.ipinfoKey, ip_address
         )
-        ipwhois_result = await get_ipwhois_result(ip_address, tmp)
+        ipwhois_result = await get_ipwhois_result(ip_address)
 
         merged_results = merge_ipinfo_ipwhois(ipinfo_result, ipwhois_result)
 
@@ -128,9 +128,32 @@ class Network(commands.Cog):
             ),
         )
 
-        await tmp.delete()
         await ctx.send(embed=e)
 
+    @command_extra(name="cloudflare", deletable=True)
+    async def _cloudflare(
+        self,
+        ctx: ContextPlus,
+        ip: DomainConverter,
+    ):
+        crimeflare_result = await get_crimeflare_result(
+            self.bot.session, str(ip)
+        )
+
+        if crimeflare_result:
+            alt_ctx = await copy_context_with(
+                ctx, content=f"{ctx.prefix}iplocalise {crimeflare_result}"
+            )
+            return await alt_ctx.command.reinvoke(alt_ctx)
+
+        await ctx.send(
+            _(
+                "Unable to collect information through CloudFlare",
+                ctx,
+                self.bot.config,
+            ).format()
+        )
+
     @command_extra(name="getheaders", aliases=["headers"], deletable=True)
     async def _getheaders(
         self, ctx: ContextPlus, ip: DomainConverter, *, user_agent: str = ""