Nathan Woodburn
eaa9ebdc54
All checks were successful
Build Docker / Build Image (push) Successful in 39s
120 lines
3.4 KiB
Python
120 lines
3.4 KiB
Python
import os
|
|
from dotenv import load_dotenv
|
|
import dns.resolver
|
|
import dns.exception
|
|
import dns.message
|
|
import discord
|
|
import json
|
|
import requests
|
|
import re
|
|
|
|
load_dotenv()
|
|
|
|
resolver = dns.resolver.Resolver()
|
|
serverIP = os.getenv('DNS_SERVER')
|
|
if not serverIP:
|
|
print("No DNS_SERVER found in .env")
|
|
exit(1)
|
|
resolver.nameservers = [serverIP]
|
|
resolver.port = int(os.getenv('DNS_PORT',"53"))
|
|
|
|
LOCAL = False
|
|
if os.getenv('LOCAL') == "True":
|
|
LOCAL = True
|
|
|
|
verified_roles = '/data/roles.json'
|
|
|
|
if LOCAL:
|
|
verified_roles = 'roles.json'
|
|
|
|
|
|
def check_name(user_id: int, name: str) -> bool:
|
|
try:
|
|
answer = resolver.resolve('_shaker._auth.' + name, 'TXT')
|
|
for rrset in answer.response.answer:
|
|
parts = rrset.to_text().split(" ")
|
|
if str(user_id) in parts[-1]:
|
|
return True
|
|
except dns.exception.DNSException as e:
|
|
print("DNS Exception")
|
|
print(e)
|
|
|
|
# Alternatively check via hsd.hns.au
|
|
try:
|
|
response = requests.get(f"https://hsd.hns.au/api/v1/nameresource/{name}")
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if 'records' not in data:
|
|
return False
|
|
for record in data['records']:
|
|
if record['type'] == 'TXT' and str(user_id) in record['txt']:
|
|
return True
|
|
|
|
except Exception as e:
|
|
print("HTTP Exception")
|
|
print(e)
|
|
|
|
return False
|
|
|
|
async def handle_role(member: discord.Member, shouldHaveRole: bool):
|
|
with open(verified_roles, 'r') as f:
|
|
roles = json.load(f)
|
|
|
|
key = str(member.guild.id)
|
|
|
|
if not key in roles:
|
|
return
|
|
|
|
role_id = roles[key]
|
|
|
|
if role_id:
|
|
guild = member.guild
|
|
role = guild.get_role(role_id)
|
|
if role and shouldHaveRole and not role in member.roles:
|
|
await member.add_roles(role)
|
|
elif role and not shouldHaveRole and role in member.roles:
|
|
await member.remove_roles(role)
|
|
|
|
|
|
async def check_member(member: discord.Member) -> bool:
|
|
if member.display_name[-1] != "/":
|
|
print("No trailing /",member.display_name,flush=True)
|
|
await handle_role(member, False)
|
|
return False
|
|
|
|
domain = member.display_name[0:-1]
|
|
name_idna = domain.strip().rstrip("/").encode("idna")
|
|
name_ascii = name_idna.decode("ascii")
|
|
parts = name_ascii.split(".")
|
|
|
|
for part in parts:
|
|
if not re.match(r'[A-Za-z0-9-_]+$', part):
|
|
try:
|
|
print("Name not verified, removing /",member.display_name,flush=True)
|
|
await member.edit(nick=member.display_name[0:-1])
|
|
except Exception as e:
|
|
print(e)
|
|
await handle_role(member, False)
|
|
return False
|
|
|
|
try:
|
|
name_rendered = name_idna.decode("idna")
|
|
except UnicodeError: # don't render invalid punycode
|
|
name_rendered = name_ascii
|
|
|
|
if check_name(member.id, name_ascii):
|
|
print("Name verified",member.display_name,flush=True)
|
|
await handle_role(member, True)
|
|
return True
|
|
|
|
try:
|
|
print("Name not verified, removing /",member.display_name,flush=True)
|
|
await member.edit(nick=member.display_name[0:-1])
|
|
except Exception as e:
|
|
print(e)
|
|
await handle_role(member, False)
|
|
return False
|
|
|
|
if __name__ == "__main__":
|
|
# Try to check name
|
|
print(check_name(481051831083073536,"xn--tya")) |