From 74307c755c4eeef2678cb7a7e8b1d788b35b82fa Mon Sep 17 00:00:00 2001
From: Romain J <romain.ordi@gmail.com>
Date: Tue, 26 Jan 2021 10:21:39 +0100
Subject: [PATCH] fix(commands|Network>iplocalise): fix issues in
7962205d162698f6344f78026b8d2df989d9dd18
---
tuxbot/cogs/Dev/dev.py | 8 +-
tuxbot/cogs/Logs/logs.py | 4 +-
tuxbot/cogs/Network/functions/exceptions.py | 4 +
tuxbot/cogs/Network/functions/utils.py | 67 ++++++++++
tuxbot/cogs/Network/locales/en-US.po | 2 +-
tuxbot/cogs/Network/locales/fr-FR.po | 4 +-
tuxbot/cogs/Network/locales/messages.pot | 2 +-
tuxbot/cogs/Network/network.py | 135 +++++++-------------
tuxbot/core/bot.py | 1 +
9 files changed, 130 insertions(+), 97 deletions(-)
create mode 100644 tuxbot/cogs/Network/functions/utils.py
diff --git a/tuxbot/cogs/Dev/dev.py b/tuxbot/cogs/Dev/dev.py
index 1c3c2a2..8773369 100644
--- a/tuxbot/cogs/Dev/dev.py
+++ b/tuxbot/cogs/Dev/dev.py
@@ -21,16 +21,16 @@ class Dev(commands.Cog, name="Dev"):
def __init__(self, bot: Tux):
self.bot = bot
- self.config: DevConfig = ConfigFile(
+ self.__config: DevConfig = ConfigFile(
str(cogs_data_path(self.bot.instance_name, "Dev") / "config.yaml"),
DevConfig,
).config
# pylint: disable=invalid-name
self.yt = YouTrack(
- self.config.url.rstrip("/") + "/youtrack/",
- login=self.config.login,
- password=self.config.password,
+ self.__config.url.rstrip("/") + "/youtrack/",
+ login=self.__config.login,
+ password=self.__config.password,
)
# =========================================================================
diff --git a/tuxbot/cogs/Logs/logs.py b/tuxbot/cogs/Logs/logs.py
index ffce6f0..218e5da 100644
--- a/tuxbot/cogs/Logs/logs.py
+++ b/tuxbot/cogs/Logs/logs.py
@@ -53,7 +53,7 @@ class Logs(commands.Cog, name="Logs"):
self._gateway_queue = asyncio.Queue(loop=bot.loop)
self.gateway_worker.start() # pylint: disable=no-member
- self.config: LogsConfig = ConfigFile(
+ self.__config: LogsConfig = ConfigFile(
str(
cogs_data_path(self.bot.instance_name, "Logs") / "config.yaml"
),
@@ -129,7 +129,7 @@ class Logs(commands.Cog, name="Logs"):
def webhook(self, log_type):
webhook = discord.Webhook.from_url(
- getattr(self.config, log_type),
+ getattr(self.__config, log_type),
adapter=discord.AsyncWebhookAdapter(self.bot.session),
)
return webhook
diff --git a/tuxbot/cogs/Network/functions/exceptions.py b/tuxbot/cogs/Network/functions/exceptions.py
index 9786340..ee7a62c 100644
--- a/tuxbot/cogs/Network/functions/exceptions.py
+++ b/tuxbot/cogs/Network/functions/exceptions.py
@@ -7,3 +7,7 @@ class RFC18(commands.UserNotFound):
class InvalidIp(commands.BadArgument):
pass
+
+
+class VersionNotFound(commands.BadArgument):
+ pass
diff --git a/tuxbot/cogs/Network/functions/utils.py b/tuxbot/cogs/Network/functions/utils.py
new file mode 100644
index 0000000..dba9730
--- /dev/null
+++ b/tuxbot/cogs/Network/functions/utils.py
@@ -0,0 +1,67 @@
+import socket
+from typing import Union, NoReturn
+
+import ipinfo
+import ipwhois
+from ipinfo.exceptions import RequestQuotaExceededError
+
+from ipwhois import Net
+from ipwhois.asn import IPASN
+
+from tuxbot.cogs.Network.functions.exceptions import VersionNotFound, RFC18
+
+
+def _(x):
+ return x
+
+
+async def get_ip(ip: str, inet: str = "") -> str:
+ if inet == "6":
+ inet = socket.AF_INET6
+ elif inet == "4":
+ inet = socket.AF_INET
+ else:
+ inet = 0
+
+ try:
+ return socket.getaddrinfo(str(ip), None, inet)[1][4][0]
+ except socket.gaierror as e:
+ raise VersionNotFound(
+ _(
+ "Impossible to collect information on this in the given "
+ "version",
+ )
+ ) from e
+
+
+def get_hostname(ip: str) -> str:
+ try:
+ return socket.gethostbyaddr(ip)[0]
+ except socket.herror:
+ return "N/A"
+
+
+def get_ipwhois_result(ip_address: str) -> Union[NoReturn, dict]:
+ try:
+ net = Net(ip_address)
+ obj = IPASN(net)
+ return obj.lookup()
+ except ipwhois.exceptions.ASNRegistryError:
+ return {}
+ except ipwhois.exceptions.IPDefinedError as e:
+ raise RFC18(
+ _(
+ "IP address {ip_address} is already defined as Private-Use"
+ " Networks via RFC 1918."
+ )
+ ) from e
+
+
+async def get_ipinfo_result(
+ apikey: str, ip_address: str
+) -> Union[NoReturn, dict]:
+ try:
+ handler = ipinfo.getHandlerAsync(apikey)
+ return (await handler.getDetails(ip_address)).all
+ except RequestQuotaExceededError:
+ return {}
diff --git a/tuxbot/cogs/Network/locales/en-US.po b/tuxbot/cogs/Network/locales/en-US.po
index 1b37c06..fba187e 100644
--- a/tuxbot/cogs/Network/locales/en-US.po
+++ b/tuxbot/cogs/Network/locales/en-US.po
@@ -33,7 +33,7 @@ msgstr ""
#: tuxbot/cogs/Network/network.py:61
#, python-brace-format
-msgid "Impossible to collect information on this ip {version}"
+msgid "Impossible to collect information on this in the given version"
msgstr ""
#: tuxbot/cogs/Network/network.py:89
diff --git a/tuxbot/cogs/Network/locales/fr-FR.po b/tuxbot/cogs/Network/locales/fr-FR.po
index 61b432a..a080354 100644
--- a/tuxbot/cogs/Network/locales/fr-FR.po
+++ b/tuxbot/cogs/Network/locales/fr-FR.po
@@ -33,8 +33,8 @@ msgstr "en v{v}"
#: tuxbot/cogs/Network/network.py:61
#, python-brace-format
-msgid "Impossible to collect information on this ip {version}"
-msgstr "Impossible de collecter des informations pour cette IP {version}"
+msgid "Impossible to collect information on this in the given version"
+msgstr "Impossible de collecter des informations pour cette IP avec la version donnée"
#: tuxbot/cogs/Network/network.py:89
#, python-brace-format
diff --git a/tuxbot/cogs/Network/locales/messages.pot b/tuxbot/cogs/Network/locales/messages.pot
index 3c60d0a..906a2cd 100644
--- a/tuxbot/cogs/Network/locales/messages.pot
+++ b/tuxbot/cogs/Network/locales/messages.pot
@@ -32,7 +32,7 @@ msgstr ""
#: tuxbot/cogs/Network/network.py:61
#, python-brace-format
-msgid "Impossible to collect information on this ip {version}"
+msgid "Impossible to collect information on this in the given version"
msgstr ""
#: tuxbot/cogs/Network/network.py:89
diff --git a/tuxbot/cogs/Network/network.py b/tuxbot/cogs/Network/network.py
index 9455c5f..5ceec21 100644
--- a/tuxbot/cogs/Network/network.py
+++ b/tuxbot/cogs/Network/network.py
@@ -1,22 +1,23 @@
import functools
import logging
-import socket
-from typing import Union, NoReturn
+
import discord
-import ipinfo
-import ipwhois
+
from discord.ext import commands
from ipinfo.exceptions import RequestQuotaExceededError
-from ipwhois import Net
-from ipwhois.asn import IPASN
+
from structured_config import ConfigFile
from tuxbot.cogs.Network.functions.converters import (
IPConverter,
IPVersionConverter,
)
-from tuxbot.cogs.Network.functions.exceptions import RFC18, InvalidIp
+from tuxbot.cogs.Network.functions.exceptions import (
+ RFC18,
+ InvalidIp,
+ VersionNotFound,
+)
from tuxbot.core.bot import Tux
from tuxbot.core.i18n import (
Translator,
@@ -27,15 +28,23 @@ from tuxbot.core.utils.functions.extra import (
command_extra,
)
from .config import NetworkConfig
+from .functions.utils import (
+ get_ip,
+ get_hostname,
+ get_ipinfo_result,
+ get_ipwhois_result,
+)
log = logging.getLogger("tuxbot.cogs.Network")
_ = Translator("Network", __file__)
class Network(commands.Cog, name="Network"):
+ _tmp: discord.Message
+
def __init__(self, bot: Tux):
self.bot = bot
- self.config: NetworkConfig = ConfigFile(
+ self.__config: NetworkConfig = ConfigFile(
str(
cogs_data_path(self.bot.instance_name, "Network")
/ "config.yaml"
@@ -44,76 +53,18 @@ class Network(commands.Cog, name="Network"):
).config
async def cog_command_error(self, ctx, error):
- if isinstance(error, (RequestQuotaExceededError, RFC18, InvalidIp)):
+ if isinstance(
+ error,
+ (RequestQuotaExceededError, RFC18, InvalidIp, VersionNotFound),
+ ):
+ if self._tmp:
+ await self._tmp.delete()
+
await ctx.send(_(str(error), ctx, self.bot.config))
# =========================================================================
# =========================================================================
- async def _get_ip(self, ctx: ContextPlus, ip: str, inet: str = "") -> str:
- inet_text = ""
-
- if inet == "6":
- inet = socket.AF_INET6
- inet_text = _("in v{v}", ctx, self.bot.config).format(v=inet)
- elif inet == "4":
- inet = socket.AF_INET
- inet_text = _("in v{v}", ctx, self.bot.config).format(v=inet)
- else:
- inet = 0
-
- try:
- return socket.getaddrinfo(str(ip), None, inet)[1][4][0]
- except socket.gaierror:
- return await ctx.send(
- _(
- "Impossible to collect information on this ip {version}".format(
- version=inet_text
- ),
- ctx,
- self.bot.config,
- )
- )
-
- @staticmethod
- def _get_hostname(ip: str) -> str:
- try:
- return socket.gethostbyaddr(ip)[0]
- except socket.herror:
- return "N/A"
-
- @staticmethod
- def get_ipwhois_result(ip_address: str) -> Union[NoReturn, dict]:
- try:
- net = Net(ip_address)
- obj = IPASN(net)
- return obj.lookup()
- except ipwhois.exceptions.ASNRegistryError:
- return {}
- except ipwhois.exceptions.IPDefinedError as e:
-
- def _(x):
- return x
-
- raise RFC18(
- _(
- "IP address {ip_address} is already defined as Private-Use"
- " Networks via RFC 1918."
- )
- ) from e
-
- async def get_ipinfo_result(
- self, ip_address: str
- ) -> Union[NoReturn, dict]:
- try:
- handler = ipinfo.getHandlerAsync(self.config.ipinfoKey)
- return (await handler.getDetails(ip_address)).all
- except RequestQuotaExceededError:
- return {}
-
- # =========================================================================
- # =========================================================================
-
@command_extra(name="iplocalise", aliases=["localiseip"], deletable=True)
async def _iplocalise(
self,
@@ -121,17 +72,19 @@ class Network(commands.Cog, name="Network"):
ip: IPConverter,
version: IPVersionConverter = "",
):
- tmp = await ctx.send(
+ self._tmp = await ctx.send(
_("*Retrieving information...*", ctx, self.bot.config),
deletable=False,
)
- ip_address = await self._get_ip(ctx, str(ip), str(version))
- ip_hostname = self._get_hostname(ip_address)
+ ip_address = await get_ip(str(ip), str(version))
+ ip_hostname = get_hostname(ip_address)
- ipinfo_result = await self.get_ipinfo_result(ip_address)
+ ipinfo_result = await get_ipinfo_result(
+ self.__config.ipinfoKey, ip_address
+ )
ipwhois_result = await self.bot.loop.run_in_executor(
- None, functools.partial(self.get_ipwhois_result, ip_address)
+ None, functools.partial(get_ipwhois_result, ip_address)
)
e = discord.Embed(
@@ -151,11 +104,18 @@ class Network(commands.Cog, name="Network"):
inline=True,
)
+ if ipwhois_result:
+ e.add_field(
+ name="RIR :",
+ value=f"```{ipwhois_result['asn_registry']}```",
+ inline=True,
+ )
+
e.add_field(
name=_("Region:", ctx, self.bot.config),
- value=f"{ipinfo_result.get('city', 'N/A')} - "
+ value=f"```{ipinfo_result.get('city', 'N/A')} - "
f"{ipinfo_result.get('region', 'N/A')} "
- f"({ipinfo_result.get('country', 'N/A')})",
+ f"({ipinfo_result.get('country', 'N/A')})```",
inline=False,
)
@@ -174,9 +134,15 @@ class Network(commands.Cog, name="Network"):
inline=True,
)
+ e.add_field(
+ name="RIR :",
+ value=f"```{ipwhois_result['asn_registry']}```",
+ inline=True,
+ )
+
e.add_field(
name=_("Region:", ctx, self.bot.config),
- value=asn_country,
+ value=f"```{asn_country}```",
inline=False,
)
@@ -184,16 +150,11 @@ class Network(commands.Cog, name="Network"):
url=f"https://www.countryflags.io/{asn_country}/shiny/64.png"
)
- if ipwhois_result:
- e.add_field(
- name="RIR :", value=ipwhois_result["asn_registry"], inline=True
- )
-
e.set_footer(
text=_("Hostname: {hostname}", ctx, self.bot.config).format(
hostname=ip_hostname
),
)
- await tmp.delete()
+ await self._tmp.delete()
await ctx.send(embed=e)
diff --git a/tuxbot/core/bot.py b/tuxbot/core/bot.py
index df720bf..ea3d762 100644
--- a/tuxbot/core/bot.py
+++ b/tuxbot/core/bot.py
@@ -45,6 +45,7 @@ packages: List[str] = [
"tuxbot.cogs.Utils",
"tuxbot.cogs.Polls",
"tuxbot.cogs.Custom",
+ "tuxbot.cogs.Network",
]