gnouseu/tuxbot-bot
4
0
Fork 0
Code Issues Pull requests Wiki Activity

fix(commands|Network>iplocalise): fix issues in 7962205d16

Browse source
This commit is contained in:
Romain J 2021-01-26 10:21:39 +01:00
parent 18310f17a0
commit 74307c755c
9 changed files with 130 additions and 97 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
tuxbot/cogs/Dev/dev.py
View file

@ -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,
)
# =========================================================================

4
tuxbot/cogs/Logs/logs.py
View file

@ -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

4
tuxbot/cogs/Network/functions/exceptions.py
View file

@ -7,3 +7,7 @@ class RFC18(commands.UserNotFound):
class InvalidIp(commands.BadArgument):
pass
class VersionNotFound(commands.BadArgument):
pass

67
tuxbot/cogs/Network/functions/utils.py Normal file
View file

@ -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 {}

2
tuxbot/cogs/Network/locales/en-US.po
View file

@ -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

4
tuxbot/cogs/Network/locales/fr-FR.po
View file

@ -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

2
tuxbot/cogs/Network/locales/messages.pot
View file

@ -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

135
tuxbot/cogs/Network/network.py
View file

@ -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)

1
tuxbot/core/bot.py
View file

@ -45,6 +45,7 @@ packages: List[str] = [
"tuxbot.cogs.Utils",
"tuxbot.cogs.Polls",
"tuxbot.cogs.Custom",
"tuxbot.cogs.Network",
]