Skip to content

Instantly share code, notes, and snippets.

@m8sec
Last active July 26, 2022 02:15
Show Gist options
  • Save m8sec/e4d8652f5c409c71c8fab992e74ede3c to your computer and use it in GitHub Desktop.
Save m8sec/e4d8652f5c409c71c8fab992e74ede3c to your computer and use it in GitHub Desktop.
Python script to monitor a Slack channel and automate task execution.
#!/usr/bin/env python3
# Author: @m8sec
import os
import threading
from sys import exit
from time import sleep
from datetime import datetime
from subprocess import getoutput
from taser.proto.http import web_request
####### Description #######
# Python script to monitor a Slack channel and automate task
# execution. As an example, this will automate the use of Nmap
# on a remote server.
#
# By default uploaded files and reports will be written to
# the local directory. This can be modified in the config
# settings below.
#
# For best results, install as a service using systemctl. This
# will allow for persistent monitoring of the Slack channel
# and prevent the need for an open SSH session on the host.
#
####### Requirements #######
# Python 3.6+
# Debian-based Linux distro (Ubuntu)
#
# Install Req:
# sudo pip3 install taser
# sudo apt install nmap
#
####### Slack Config Settings #######
executing_users = ['my_user'] # Authorized users
slack_token = '' # Slack App token.
slack_channel = 'nmap' # Slack channel to monitor.
slack_checkin = 10 # Checkin time (seconds).
nmap_cmd = 'nmap -sS --min-rate 500 -T 4 -open ' # Default nmap command to execute.
nmap_report = 'slackexec_report.xml' # Default nmap report name.
default_upload = os.path.dirname(os.path.realpath(__file__)) # Default dir to upload files.
####### Slack API Class #######
class SlackAPI():
@staticmethod
def chat_postMsg(msg, channel_id):
'''Post Message to channel as bot'''
data = {
'channel': channel_id,
'text': str(msg).replace('"', '\"')
}
header = {'Authorization': 'Bearer '+slack_token,
'Content-type': 'application/json'}
return web_request('https://slack.com/api/chat.postMessage', method='POST', headers=header, json=data)
@staticmethod
def conversations_list():
'''List Channels'''
url='https://slack.com/api/conversations.list'
return web_request(url, method='GET', headers={'Authorization': 'Bearer ' + slack_token})
@staticmethod
def conversations_history(channel_id, limit=1):
'''Get chat history'''
url = 'https://slack.com/api/conversations.history?channel={}&limit={}'.format(channel_id, str(limit))
return web_request(url, method='GET', headers={'Authorization': 'Bearer ' + slack_token})
@staticmethod
def file_list(channel_id, limit=1):
'''List files uploaded a channel and return private download link'''
links = []
try:
url = 'https://slack.com/api/files.list?channel={}&count={}'.format(channel_id, str(limit))
resp = web_request(url, method='GET', headers={'Authorization': 'Bearer ' + slack_token})
for x in resp.json()['files']:
links.append({
'url' : x['url_private_download'],
'name': x['name'],
'time': x['timestamp']
})
except:
pass
return links
@staticmethod
def file_upload(filename, channel_id, comment:str=''):
'''Upload file to channel, full file path required'''
with open(filename, 'rb') as f:
file_contents = f.read()
data = {
'channels': channel_id,
'content' : file_contents,
'filename' : os.path.basename(filename),
'initial_comment' : comment,
}
header = {'Authorization': 'Bearer '+slack_token,
'Content-type': 'application/x-www-form-urlencoded'}
return web_request('https://slack.com/api/files.upload', method='POST', headers=header, data=data, debug=True)
@staticmethod
def users_list():
'''List workspace users'''
url = 'https://slack.com/api/users.list'
return web_request(url, method='GET', headers={'Authorization': 'Bearer ' + slack_token})
@staticmethod
def getUserName(user_id):
'''Get slack username by ID value'''
try:
for member in SlackAPI.users_list().json()['members']:
if member['id'] == user_id:
return member['name']
except Exception as e:
pass
return "n/a"
@staticmethod
def getChannelID(name):
'''Get channel ID by name'''
try:
resp = SlackAPI.conversations_list()
for x in resp.json()['channels']:
if x['name'].lower() == name.lower():
return x['id']
except:
return False
return False
@staticmethod
def download_file(source, output):
f = open(output, 'wb+')
f.write(web_request(source, headers={'Authorization': 'Bearer ' + slack_token}, timeout=5, debug=True).content)
f.close()
####### Command Execution in new thread #######
class SlackExec(threading.Thread):
def __init__(self, cmd, channel, outfile):
threading.Thread.__init__(self)
self.daemon=True
self.cmd = cmd
self.channel = channel
self.report = outfile
def run(self):
self.output = getoutput(self.cmd)
self.sendResults(self.report)
def sendResults(self, filename):
if os.path.exists(filename):
SlackAPI.file_upload(filename, self.channel, comment='')
else:
SlackAPI.chat_postMsg('Error - Report file not found.', self.channel)
####### Slack Monitor #######
class SlackBot():
def __init__(self, channel, checkin=5):
self.running = True
self.channel = channel
self.checkin = checkin
self.last_file = ''
self.nmap = nmap_cmd
self.report = nmap_report
self.exe_users = executing_users
def get_help(self):
help = "Slack Automation\n{}\n".format('-'*25)
help += "help Return this menu.\n"
help += "status Get current status of bot.\n"
help += "shutdown Shutdown Bot on remote server.\n"
help += "checkin [#] Change bot check-in time (seconds).\n"
help += "channel [name] Change bot's monitoring channel.\n"
help += "update [nmap cmd] Update Nmap command & arguments.\n"
help += "add-user [slack user] Add authorized user for execution.\n"
help += "nmap [scope|file] Execute nmap scan on the scope.\n"
return '```{}```'.format(help)
def get_slackCMD(self, channel):
channel_id = SlackAPI.getChannelID(channel)
r = SlackAPI.conversations_history(channel_id).json()
return {
'channel' : channel,
'channel_id': channel_id,
'user' : SlackAPI.getUserName(r['messages'][0]['user']),
'type' : r['messages'][0]['type'],
'cmd' : r['messages'][0]['text'].strip()
}
def uploadFile(self, msg):
sleep(5) # give Slack time to process file
file = SlackAPI.file_list(msg['channel_id'])
if file[0]['url'] == self.last_file:
return False
try:
t = msg['cmd'].strip().split(' ')
upload_dir = t[-1] if os.path.isdir(t[-1]) else default_upload
upload_dir = upload_dir if upload_dir.endswith('/') else upload_dir+'/'
upload_dir = dupCheck(upload_dir+file[0]['name'])
SlackAPI.download_file(file[0]['url'], upload_dir)
self.last_file = file[0]['url']
return upload_dir
except:
SlackAPI.chat_postMsg('Upload failed, check inputs and try again.', msg['channel_id'])
return False
def Executioner(self, msg):
if msg['user'].lower() not in self.exe_users:
return
if msg['cmd'] == 'shutdown':
m = 'Shutdown initiated by {} @ {}'.format(msg['user'], datetime.now().strftime('%Y/%m/%d %H:%M'))
SlackAPI.chat_postMsg(m, msg['channel_id'])
exit(0)
elif msg['cmd'] == 'help':
SlackAPI.chat_postMsg(self.get_help(), msg['channel_id'])
elif msg['cmd'].lower().startswith('update'):
self.nmap = msg['cmd'].strip('update ') + ' '
SlackAPI.chat_postMsg('Nmap Command updated: {}'.format(self.nmap), msg['channel_id'])
elif msg['cmd'].lower().startswith('add-user'):
self.exe_users.append(msg['cmd'].strip('add-user ').strip())
SlackAPI.chat_postMsg('Executing users updated.', msg['channel_id'])
elif msg['cmd'] == 'status':
status = 'Channel : {}\n'.format(self.channel)
status+= 'Checkin : {} sec.\n'.format(str(self.checkin))
status+= 'Source IP : {}\n'.format(get_externalIP())
status+= 'Upload Dir : {}\n'.format(default_upload)
status+= 'Nmap Cmd : {}\n'.format(self.nmap)
status+= 'Authorized Users : {}\n'.format(self.exe_users)
SlackAPI.chat_postMsg('```{}```'.format(status), msg['channel_id'])
elif msg['cmd'].startswith('checkin'):
try:
cmd, time = msg['cmd'].split(' ')
time = int(time)
SlackAPI.chat_postMsg('```Changing check-in time: {} => {} (seconds)```'.format(self.checkin, time),msg['channel_id'])
self.checkin = time
return
except:
pass
SlackAPI.chat_postMsg('Invalid input, expecting integer value.', msg['channel_id'])
elif msg['cmd'].startswith('channel'):
try:
cmd, channel = msg['cmd'].split(' ')
if SlackAPI.getChannelID(channel):
self.channel = channel
SlackAPI.chat_postMsg('```Monitoring channel changed to: {}```'.format(channel), msg['channel_id'])
return
except:
pass
SlackAPI.chat_postMsg('Invalid input, check channel and try again.', msg['channel_id'])
elif msg['cmd'].lower().startswith('nmap'):
report_name = dupCheck(self.report)
if msg['cmd'].lower() == 'nmap':
scope = self.uploadFile(msg)
cmd = self.nmap + "-iL " + scope + " -oX " + report_name + ' > /dev/null 2>&1'
else:
scope = msg['cmd'].strip('nmap ').strip()
if scope.startswith(('<https://','<http://')):
scope = scope.split('|')[1].split('>')[0]
cmd = self.nmap + scope + " -oX " + report_name + ' > /dev/null 2>&1'
if scope:
SlackAPI.chat_postMsg("Starting nmap against scope: {}".format(scope), msg['channel_id'])
SlackExec(cmd, msg['channel_id'], report_name).start()
else:
SlackAPI.chat_postMsg('Slack Automation - Invalid Input.', msg['channel_id'])
else:
SlackAPI.chat_postMsg('Slack Automation - Invalid Input.', msg['channel_id'])
def startLoop(self):
while True:
try:
self.Executioner(self.get_slackCMD(self.channel))
except KeyboardInterrupt:
exit(0)
except:
pass
sleep(self.checkin)
####### Support Function(s) #######
def get_externalIP():
try:
return web_request('http://ident.me').text
except:
return 'N/A'
def dupCheck(filename):
'''Check for duplicate files and rename'''
c = 1
f = filename
tmp = f
while os.path.exists(tmp):
tmp = f+"_"+str(c)
c +=1
return tmp
####### Entry Point #######
if __name__ == '__main__':
SlackBot(slack_channel, slack_checkin).startLoop()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment