LimnoriaPlugins/SnoParser
6
0
Fork 1
Code Issues 9 Pull Requests 1 Projects Releases Wiki Activity

regex-cleanup #19

Merged
pratyush merged 19 commits from regex-cleanup into master 2023-05-03 13:01:49 +02:00
Conversation 0 Commits 19 Files Changed 5 +176 -55
4 changed files with 176 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 78161bda7f - Show all commits

BIN
__pycache__/config.cpython-38.pyc
View File

Binary file not shown.

BIN
__pycache__/plugin.cpython-38.pyc
View File

Binary file not shown.

96
config.py
View File

@ -61,6 +61,67 @@ conf.registerGlobalValue(SnoParser, 'AutoVhost',
conf.registerGlobalValue(SnoParser, 'preventHighlight',
registry.Boolean(True, ("""Toggles in channel highlights with ZWSP""")))
conf.registerGlobalValue(SnoParser, 'debug',
registry.Boolean('false',
"""
SnoParser: Verbose output. Note: there is a seperate debug option for the `whois` client.
"""
, private=True
))
###
# REDIS related settings below:
###
conf.registerGroup(SnoParser, 'redis')
conf.registerGlobalValue(SnoParser.redis, 'db1',
registry.Integer('1',
"""
Redis: Database number for counting of NICKNAMES.
"""
, private=True
))
conf.registerGlobalValue(SnoParser.redis, 'db2',
registry.Integer('2',
"""
Redis: Database number for counting of IP ADDRESSES.
"""
, private=True
))
conf.registerGlobalValue(SnoParser.redis, 'host',
registry.String('127.0.0.1',
"""
Redis: IP address or hostname.
"""
, private=True
))
conf.registerGlobalValue(SnoParser.redis, 'port',
registry.Integer('6379',
"""
Redis: Port.
"""
, private=True
))
conf.registerGlobalValue(SnoParser.redis, 'username',
registry.String('',
"""
Redis: Username. This is optional and has not been tested. It is recommended to perform initial tests on a local instance with no username and password.
"""
, private=True
))
conf.registerGlobalValue(SnoParser.redis, 'password',
registry.String('',
"""
Redis: Password. This is optional and has not been tested. It is recommended to perform initial tests on a local instance with no username and password.
"""
))
conf.registerGlobalValue(SnoParser.redis, 'timeout',
registry.Integer('5',
"""
Redis: Socket Timeout. The developer does not know what to recommend here, but `5` seems to yield good results.
"""
))
###
## WHOIS related settings below:
###
@ -80,43 +141,10 @@ conf.registerGlobalValue(SnoParser.whois, 'sample',
, private=True
))
conf.registerGroup(SnoParser.whois, 'redis')
conf.registerGlobalValue(SnoParser.whois.redis, 'host',
registry.String('127.0.0.1',
"""
Redis: IP address or hostname.
"""
, private=True
))
conf.registerGlobalValue(SnoParser.whois.redis, 'port',
registry.Integer('6379',
"""
Redis: Port.
"""
, private=True
))
conf.registerGlobalValue(SnoParser.whois.redis, 'username',
registry.String('',
"""
Redis: Username. This is optional and has not been tested. It is recommended to perform initial tests on a local instance with no username and password.
"""
, private=True
))
conf.registerGlobalValue(SnoParser.whois.redis, 'password',
registry.String('',
"""
Redis: Password. This is optional and has not been tested. It is recommended to perform initial tests on a local instance with no username and password.
"""
))
conf.registerGlobalValue(SnoParser.whois.redis, 'db',
registry.Integer('0',
"""
Redis: Database number. It is recommended to use a dedicated, isolated, Redis instance for SnoParser instead of using an existing one with a different database.
"""
))
conf.registerGlobalValue(SnoParser.whois.redis, 'timeout',
registry.Integer('5',
"""
Redis: Socket Timeout. The developer does not know what to recommend here, but `5` seems to yield good results.
Redis: Database number for WHOIS query caching.
"""
))

135
plugin.py
View File

@ -53,25 +53,60 @@ class SnoParser(callbacks.Plugin):
"""Parses the Server Notices from ErgoIRCd"""
threaded = True
def redis_connect(self) -> redis.client.Redis:
def redis_connect_whois(self) -> redis.client.Redis:
try:
redis_client = redis.Redis(
host = self.registryValue('whois.redis.host'),
port = self.registryValue('whois.redis.port'),
password = self.registryValue('whois.redis.password'),
username = self.registryValue('whois.redis.username'),
redis_client_whois = redis.Redis(
host = self.registryValue('redis.host'),
port = self.registryValue('redis.port'),
password = self.registryValue('redis.password'),
username = self.registryValue('redis.username'),
db = self.registryValue('whois.redis.db'),
socket_timeout = int(self.registryValue('whois.redis.timeout'))
socket_timeout = int(self.registryValue('redis.timeout'))
)
ping = redis_client.ping()
ping = redis_client_whois.ping()
if ping is True:
return redis_client
return redis_client_whois
except redis.AuthenticationError:
print("Could not authenticate to Redis backend.")
def redis_connect_nicks(self) -> redis.client.Redis:
try:
redis_client_nicks = redis.Redis(
host = self.registryValue('redis.host'),
port = self.registryValue('redis.port'),
password = self.registryValue('redis.password'),
username = self.registryValue('redis.username'),
db = self.registryValue('redis.db1'),
socket_timeout = int(self.registryValue('redis.timeout'))
)
ping = redis_client_nicks.ping()
if ping is True:
return redis_client_nicks
except redis.AuthenticationError:
print("Could not authenticate to Redis backend.")
def redis_connect_ips(self) -> redis.client.Redis:
try:
redis_client_ips = redis.Redis(
host = self.registryValue('redis.host'),
port = self.registryValue('redis.port'),
password = self.registryValue('redis.password'),
username = self.registryValue('redis.username'),
db = self.registryValue('redis.db2'),
socket_timeout = int(self.registryValue('redis.timeout'))
)
ping = redis_client_ips.ping()
if ping is True:
return redis_client_ips
except redis.AuthenticationError:
print("Could not authenticate to Redis backend.")
def __init__(self, irc):
super().__init__(irc)
self.redis_client = self.redis_connect()
self.redis_client_whois = self.redis_connect_whois()
self.redis_client_nicks = self.redis_connect_nicks()
self.redis_client_ips = self.redis_connect_ips()
def whois_fresh(self, sourceip: str) -> dict:
"""Data from cache."""
@ -96,17 +131,17 @@ class SnoParser(callbacks.Plugin):
def whois_get_cache(self, key: str) -> str:
"""Data from Redis."""
k = self.redis_client.get(key)
k = self.redis_client_whois.get(key)
# self = SnoParser()
# val = self.redis_client.get(key)
# val = self.redis_client_whois.get(key)
val = k
return val
def whois_set_cache(self, key: str, value: str) -> bool:
"""Data to Redis."""
state = self.redis_client.setex(key, timedelta(seconds=3600), value=value,)
state = self.redis_client_whois.setex(key, timedelta(seconds=3600), value=value,)
return state
def whois_run(self, sourceip: str) -> dict:
@ -141,19 +176,76 @@ class SnoParser(callbacks.Plugin):
print(data)
return data
def query(self, irc, msg, args, ipaddress):
def nick_run(self, nickname: str) -> dict:
"""Tracks nicknames"""
data = self.redis_client_nicks.get(nickname)
if data is not None:
if self.registryValue('debug'):
print("SNOPARSER DEBUG - NICK_RUN, SEEN: TRUE")
print(nickname)
print(data)
self.redis_client_nicks.incrby(nickname,amount=1)
if data:
decoded = data.decode('utf-8')
return decoded
else:
return 0
else:
if self.registryValue('debug'):
print("SNOPARSER DEBUG _ NICK_RUN, SEEN: FALSE")
print(nickname)
print(data)
self.redis_client_nicks.set(nickname,value='1')
if data:
decoded = data.decode('utf-8')
return decoded
else:
return 0
def ip_run(self, ipaddress: str) -> dict:
"""Tracks IP addresses"""
data = self.redis_client_ips.get(ipaddress)
if data is not None:
if self.registryValue('debug'):
print("SNOPARSER DEBUG - IP_RUN, SEEN: TRUE")
print(ipaddress)
print(data)
self.redis_client_ips.incrby(ipaddress,amount=1)
if data:
decoded = data.decode('utf-8')
return decoded
else:
return 0
else:
if self.registryValue('debug'):
print("SNOPARSER DEBUG _ IP_RUN, SEEN: FALSE")
print(ipaddress)
print(data)
self.redis_client_ips.set(ipaddress,value='1')
if data:
decoded = data.decode('utf-8')
return decoded
else:
return 0
def ipquery(self, irc, msg, args, ipaddress):
"""<IP address>
Queries the cache for an address.
"""
data = self.whois_get_cache(key=ipaddress)
decoded = data.decode('utf-8')
ttl = self.redis_client.ttl(ipaddress)
ttl = self.redis_client_whois.ttl(ipaddress)
count = self.redis_client_ips.get(ipaddress)
print('SnoParser manual query: ', data, ' ', ttl)
irc.reply(f'{decoded} - Remaining: {ttl}s')
irc.reply(f'{decoded} - Count: {count} - Remaining: {ttl}s')
query = wrap(query, ['anything'])
ipquery = wrap(ipquery, ['anything'])
def doNotice(self, irc, msg):
@ -173,13 +265,14 @@ class SnoParser(callbacks.Plugin):
else:
ip = couple.group(4)
realname = couple.group(5)
ip_seen = 0
nick_seen = 0
whoisout = self.whois_run(sourceip=ip)
ip_seen = self.ip_run(ipaddress=ip)
nick_seen = self.nick_run(nickname=nickname)
whois = self.whois_run(sourceip=ip)
DictFromSnotice = {'notice': 'connect', 'nickname': nickname, 'username': username, 'host': host, 'ip': ip, 'realname': realname, 'ipCount': ip_seen, 'nickCount': nick_seen}
#repl = f"\x02\x1FNOTICE: {DictFromSnotice['notice']} \x0F\x11\x0303==>>\x0F \x02Nick:\x0F {DictFromSnotice['nickname']} \x02Username:\x0F {DictFromSnotice['username']} \x02Hostname:\x0F {DictFromSnotice['host']} \x02IP:\x0F {DictFromSnotice['ip']} \x02Realname:\x0F {DictFromSnotice['realname']} \x02IPcount:\x0F {DictFromSnotice['ipCount']} \x02NickCount:\x0F {DictFromSnotice['nickCount']}"
repl = f"\x02\x1F{DictFromSnotice['notice']} \x0F\x11\x0303==>>\x0F \x02Nick:\x0F {DictFromSnotice['nickname']} \x02Username:\x0F {DictFromSnotice['username']} \x02Hostname:\x0F {DictFromSnotice['host']} \x02IP:\x0F {DictFromSnotice['ip']} {whoisout} \x02Realname:\x0F {DictFromSnotice['realname']} \x02IPcount:\x0F {DictFromSnotice['ipCount']} \x02NickCount:\x0F {DictFromSnotice['nickCount']}"
repl = f"\x02\x1F{DictFromSnotice['notice']} \x0F\x11\x0303==>>\x0F \x02Nick:\x0F {DictFromSnotice['nickname']} \x02Username:\x0F {DictFromSnotice['username']} \x02Hostname:\x0F {DictFromSnotice['host']} \x02IP:\x0F {DictFromSnotice['ip']} {whois} \x02Realname:\x0F {DictFromSnotice['realname']} \x02IPcount:\x0F {DictFromSnotice['ipCount']} \x02NickCount:\x0F {DictFromSnotice['nickCount']}"
self._sendSnotice(irc, msg, repl)
if 'XLINE' in text and 'temporary' in text: