Networking

Bases: Cog

Class that holds networking commands This class extends commands.Cog from discord.

Parameters:
  • bot

    Discord API client

  • logger

    Logger object for logging purposes

Attributes:
  • bot

    Discord API client

  • logger

    Logger object for logging purposes

  • hidden (bool) –

    Attribute that determines if this list of command should show in the help command or not. If false, will show in help.

  • __cog_name__ (str) –

    Command designation for the help command

Source code in /Users/caoma/Documents/Programming/GIT/SkyWizz/skywizz/cogs/networking_commands.py
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
class Networking(commands.Cog):
    """
        Class that holds networking commands
        This class extends `commands.Cog` from discord.

        Args:
            bot: Discord API client
            logger: Logger object for logging purposes

        Attributes:
            bot: Discord API client
            logger: Logger object for logging purposes
            hidden (bool): Attribute that determines if this list of
                     command should show in the help command or not.
                     If `false`, will show in help.
            __cog_name__ (str): Command designation for the help command
    """

    def __init__(self, bot, logger):
        self.bot = bot
        self.logger = logger
        self.hidden = False
        self.__cog_name__ = 'Networking'
        self.logger.info(f"Loaded {self.__cog_name__}")

    @commands.cooldown(2, 60, commands.BucketType.user)
    @commands.command(name='ping', aliases=['pong'])
    async def ping(self, ctx):
        """
        Command that shows the round-trip time of the bot's connection to Discord.

        Usage:
            `ping`
        """
        # convert to milliseconds and round to whole number
        rtt = round(self.bot.latency * 1000)

        embed = embd.newembed(title="Pong!",
                              description=f"Round-trip time: {rtt} ms")

        await ctx.send(embed=embed)

    @commands.cooldown(2, 30, commands.BucketType.user)
    @commands.command(name='whois')
    async def whois_command(self, ctx, *, domain: str):
        """
        Command that performs a WHOIS lookup for the specified domain.

        Args:
            domain: The domain to lookup.

        Example:
            `!whois google.com`

        Usage:
            `whois <domain>`
        """
        # Display processing message
        processing_embed = embd.newembed(title='Processing...', description='',
                                         color=0xffd700)
        processing_embed.add_field(name='Message',
                                   value=f'Performing WHOIS lookup for `{domain}`...'
                                         'please wait...')
        processing_message = await ctx.send(embed=processing_embed)

        try:
            whois_data = await tools.whois_lookup(domain)
            # Limit the output to only the relevant fields
            if whois_data['country'].lower() != 'n/a':
                flag_emoji = emoji.emojize(f":flag_{whois_data['country'].lower()}:")
            else:
                flag_emoji = ''

            embed = embd.newembed(title=f"WHOIS lookup for {domain}",
                                  description='',
                                  color=0x7289DA)
            embed.add_field(name="Name", value=whois_data['name'],
                            inline=False)
            embed.add_field(name="Country", value=f"{flag_emoji} "
                                                  f"{whois_data['country']}",
                            inline=False)
            embed.add_field(name="Registrar", value=whois_data['registrar'],
                            inline=False)
            embed.add_field(name="Creation Date",
                            value=whois_data['creation_date'], inline=False)
            embed.add_field(name="Expiration Date",
                            value=whois_data['expiration_date'], inline=False)
            embed.add_field(name="Updated Date",
                            value=whois_data['last_updated'], inline=False)
            embed.add_field(name="Name Servers",
                            value="\n".join(whois_data['name_servers']),
                            inline=False)
            embed.set_footer(text=f'Requested by {ctx.author.display_name}')
            await processing_message.edit(content=None, embed=embed)
        except Exception as e:
            error_embed = skywizz.specific_error(f"An error occurred while "
                                                 f"performing a WHOIS lookup "
                                                 f"for `{domain}`.")
            error_embed.add_field(name="Error", value=str(e))
            await processing_message.edit(content=None, embed=error_embed)

    @commands.cooldown(2, 30, commands.BucketType.user)
    @commands.command(name='traceroute', aliases=['tr'])
    async def traceroute_command(self, ctx, *args):
        """
        Traceroute command that performs a traceroute to a given host

        Args:
            *args: The IP address or domain name to trace route to

        Example:
            `!traceroute 8.8.8.8`
            `!traceroute google.com`

        Usage:
            `traceroute <host>`
        """
        if not args:
            # Display error message if traceroute command failed
            error_embed = skywizz.specific_error('Please provide a host to '
                                                 'traceroute to')
            await ctx.send(embed=error_embed)
            return

        host = args[0]

        # Display processing message
        processing_embed = embd.newembed(title='Processing...',
                                         description='',
                                         color=0xffd700)
        processing_embed.add_field(name='Message',
                                   value='Performing traceroute, '
                                         'please wait...')
        processing_message = await ctx.send(embed=processing_embed)

        try:
            # Run traceroute command with timeout of 30 seconds
            traceroute_result = subprocess.check_output(
                ['traceroute', '-m', '15', '-n', '-q', '1', '-w', '2', host],
                timeout=30, universal_newlines=True)
        except subprocess.CalledProcessError as e:
            # Display error message if traceroute command failed
            error_embed = skywizz.specific_error(f'Traceroute failed with '
                                                 f'error: {e}')
            await processing_message.edit(embed=error_embed)
            return
        except subprocess.TimeoutExpired:
            # Display message if traceroute command took too long
            timeout_embed = skywizz.specific_error('Traceroute took too long '
                                                   'and was cancelled')
            await processing_message.edit(embed=timeout_embed)
            return

        # Traceroute successful, create and send embed with results
        results_embed = embd.newembed(title=f'Traceroute to {host}',
                                      description='',
                                      color=0x00ff00)
        results_embed.add_field(name='Results:',
                                value=f'```{traceroute_result}```')
        await processing_message.edit(embed=results_embed)

ping(ctx) async

Command that shows the round-trip time of the bot's connection to Discord.

Usage

ping

/Users/caoma/Documents/Programming/GIT/SkyWizz/skywizz/cogs/networking_commands.py
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@commands.cooldown(2, 60, commands.BucketType.user)
@commands.command(name='ping', aliases=['pong'])
async def ping(self, ctx):
    """
    Command that shows the round-trip time of the bot's connection to Discord.

    Usage:
        `ping`
    """
    # convert to milliseconds and round to whole number
    rtt = round(self.bot.latency * 1000)

    embed = embd.newembed(title="Pong!",
                          description=f"Round-trip time: {rtt} ms")

    await ctx.send(embed=embed)

traceroute_command(ctx, *args) async

Traceroute command that performs a traceroute to a given host

Parameters:
  • *args

    The IP address or domain name to trace route to

Example

!traceroute 8.8.8.8 !traceroute google.com

Usage

traceroute <host>

/Users/caoma/Documents/Programming/GIT/SkyWizz/skywizz/cogs/networking_commands.py
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
@commands.cooldown(2, 30, commands.BucketType.user)
@commands.command(name='traceroute', aliases=['tr'])
async def traceroute_command(self, ctx, *args):
    """
    Traceroute command that performs a traceroute to a given host

    Args:
        *args: The IP address or domain name to trace route to

    Example:
        `!traceroute 8.8.8.8`
        `!traceroute google.com`

    Usage:
        `traceroute <host>`
    """
    if not args:
        # Display error message if traceroute command failed
        error_embed = skywizz.specific_error('Please provide a host to '
                                             'traceroute to')
        await ctx.send(embed=error_embed)
        return

    host = args[0]

    # Display processing message
    processing_embed = embd.newembed(title='Processing...',
                                     description='',
                                     color=0xffd700)
    processing_embed.add_field(name='Message',
                               value='Performing traceroute, '
                                     'please wait...')
    processing_message = await ctx.send(embed=processing_embed)

    try:
        # Run traceroute command with timeout of 30 seconds
        traceroute_result = subprocess.check_output(
            ['traceroute', '-m', '15', '-n', '-q', '1', '-w', '2', host],
            timeout=30, universal_newlines=True)
    except subprocess.CalledProcessError as e:
        # Display error message if traceroute command failed
        error_embed = skywizz.specific_error(f'Traceroute failed with '
                                             f'error: {e}')
        await processing_message.edit(embed=error_embed)
        return
    except subprocess.TimeoutExpired:
        # Display message if traceroute command took too long
        timeout_embed = skywizz.specific_error('Traceroute took too long '
                                               'and was cancelled')
        await processing_message.edit(embed=timeout_embed)
        return

    # Traceroute successful, create and send embed with results
    results_embed = embd.newembed(title=f'Traceroute to {host}',
                                  description='',
                                  color=0x00ff00)
    results_embed.add_field(name='Results:',
                            value=f'```{traceroute_result}```')
    await processing_message.edit(embed=results_embed)

whois_command(ctx, *, domain) async

Command that performs a WHOIS lookup for the specified domain.

Parameters:
  • domain (str) –

    The domain to lookup.

Example

!whois google.com

Usage

whois <domain>

/Users/caoma/Documents/Programming/GIT/SkyWizz/skywizz/cogs/networking_commands.py
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@commands.cooldown(2, 30, commands.BucketType.user)
@commands.command(name='whois')
async def whois_command(self, ctx, *, domain: str):
    """
    Command that performs a WHOIS lookup for the specified domain.

    Args:
        domain: The domain to lookup.

    Example:
        `!whois google.com`

    Usage:
        `whois <domain>`
    """
    # Display processing message
    processing_embed = embd.newembed(title='Processing...', description='',
                                     color=0xffd700)
    processing_embed.add_field(name='Message',
                               value=f'Performing WHOIS lookup for `{domain}`...'
                                     'please wait...')
    processing_message = await ctx.send(embed=processing_embed)

    try:
        whois_data = await tools.whois_lookup(domain)
        # Limit the output to only the relevant fields
        if whois_data['country'].lower() != 'n/a':
            flag_emoji = emoji.emojize(f":flag_{whois_data['country'].lower()}:")
        else:
            flag_emoji = ''

        embed = embd.newembed(title=f"WHOIS lookup for {domain}",
                              description='',
                              color=0x7289DA)
        embed.add_field(name="Name", value=whois_data['name'],
                        inline=False)
        embed.add_field(name="Country", value=f"{flag_emoji} "
                                              f"{whois_data['country']}",
                        inline=False)
        embed.add_field(name="Registrar", value=whois_data['registrar'],
                        inline=False)
        embed.add_field(name="Creation Date",
                        value=whois_data['creation_date'], inline=False)
        embed.add_field(name="Expiration Date",
                        value=whois_data['expiration_date'], inline=False)
        embed.add_field(name="Updated Date",
                        value=whois_data['last_updated'], inline=False)
        embed.add_field(name="Name Servers",
                        value="\n".join(whois_data['name_servers']),
                        inline=False)
        embed.set_footer(text=f'Requested by {ctx.author.display_name}')
        await processing_message.edit(content=None, embed=embed)
    except Exception as e:
        error_embed = skywizz.specific_error(f"An error occurred while "
                                             f"performing a WHOIS lookup "
                                             f"for `{domain}`.")
        error_embed.add_field(name="Error", value=str(e))
        await processing_message.edit(content=None, embed=error_embed)