malias/malias.py

#!/usr/bin/python3
import sys
import sqlite3
from pathlib import Path
from sqlite3 import Error
import urllib.request
import json
import logging
import argparse
import os
import time
from rich import print
from datetime import datetime
from string import ascii_letters, digits
from argparse import RawTextHelpFormatter
from operator import itemgetter
import httpx

# Info pages for dev
# https://mailcow.docs.apiary.io/#reference/aliases/get-aliases/get-aliases
# https://demo.mailcow.email/api/#/Aliases
# HTTPx ref -> https://www.python-httpx.org/

homefilepath = Path.home()
filepath = homefilepath.joinpath('.config/malias2')
database = filepath.joinpath('malias2.db')
logfile = filepath.joinpath('malias2.log')
Path(filepath).mkdir(parents=True, exist_ok=True)
logging.basicConfig(filename=logfile,level=logging.INFO,format='%(message)s')
logging.getLogger("httpx").setLevel(logging.ERROR)
app_version = '2.0'
db_version = '2.0.0'
footer = 'malias version %s'%(app_version)

def get_latest_release():
	version = 'N/A'
	req = httpx.get('https://gitlab.pm/api/v1/repos/rune/malias/releases/latest',
		headers={"Content-Type": "application/json"}
		)
	version = req.json()['tag_name']
	return version


def release_check():
	latest_release = get_latest_release()
	if app_version != latest_release:
		print('[b]New version available @ [i]https://iurl.no/malias[/b][/i]')
	else:
		print ('You have the the latest version. Version: %s' %(app_version))



def connect_database():
	Path(filepath).mkdir(parents=True, exist_ok=True)
	conn = None
	try:
		conn = sqlite3.connect(database)
	except Error as e:
		logging.error(time.strftime("%Y-%m-%d %H:%M") + ' - Error : ' + str(e))
		print(e)
	finally:
		if conn:
			c = conn.cursor()
			c.execute('''CREATE TABLE IF NOT EXISTS settings (
							id INTEGER NOT NULL PRIMARY KEY,
							first_run INTEGER,
							server TEXT,
							apikey TEXT NOT NULL,
							data_copy INTEGER
						)''')
			c.execute('''CREATE TABLE IF NOT EXISTS aliases
						(id integer NOT NULL PRIMARY KEY,
						alias text NOT NULL,
						goto text NOT NULL,
						created text NOT NULL)''')
			c.execute('''CREATE TABLE IF NOT EXISTS timedaliases
						(id integer NOT NULL PRIMARY KEY,
						alias text NOT NULL,
						goto text NOT NULL,
						validity text NOT NULL)''')
			c.execute('''CREATE TABLE IF NOT EXISTS dbversion
					(version integer NOT NULL DEFAULT 0)''')
			c.execute('''CREATE TABLE IF NOT EXISTS timedaliases
						(id integer NOT NULL PRIMARY KEY,
						alias text NOT NULL,
						goto text NOT NULL,
						validity text NOT NULL)''')

			conn.commit()
			first_run(conn)
			return conn

def first_run(conn):
	now = datetime.now().strftime("%m-%d-%Y %H:%M")
	cursor = conn.cursor()
	cursor.execute('SELECT count(*) FROM settings')
	count = cursor.fetchone()[0]
	if count == 0:
		logging.error(now + ' - First run!')
		cursor.execute('INSERT INTO settings values(?,?,?,?,?)', (0, 1, 'dummy.server','DUMMY_KEY',0))
		cursor.execute('INSERT INTO dbversion values(?)', (db_version,))
		conn.commit()
		return None


def get_settings(kind):
	now = datetime.now().strftime("%m-%d-%Y %H:%M")
	cursor = conn.cursor()
	cursor.execute('SELECT * FROM settings')
	data = cursor.fetchall()
	first_run_status = data[0][1] # pyright: ignore
	server = data[0][2] # pyright: ignore
	api_key = data[0][3] # pyright: ignore
	copy_status = data[0][4] # pyright: ignore
	if kind == 'connection':
		if server == 'dummy.server' or api_key =='DUMMY_KEY':
			print('Error: No mailcow server or API key registered. Please add with [b]malias -s [i]your.server APIKEY[/i][/b]')
			exit(0)
		else:
			connection = {'key': api_key, 'server':server}
			req = httpx.get('https://'+connection['server']+'/api/v1/get/domain/0',
				headers={"Content-Type": "application/json",
						'X-API-Key': connection['key']
						}
						)
			data=req.json()
			code = str(req)
			if code.find('200') != -1:
				return connection # pyright: ignore
			else:
				logging.error(now + ' - Error : Server returned error: %s,  ' %(data['msg']))
				print('\n [b red]Error[/b red] : Server returned error [b]%s[/b]\n\n' %(data['msg']))
				exit(0)
	if kind == 'first_run_status':
		return first_run_status
	if kind == 'copy_status':
		return copy_status


def get_last_timed(username):
	connection = get_settings('connection')
	req = httpx.get('https://'+connection['server']+'/api/v1/get/time_limited_aliases/%s' %username,
		headers={"Content-Type": "application/json",
		'X-API-Key': connection['key']
		}
		)
	data = req.json()
	data.sort(key = itemgetter('validity'), reverse=True)
	return(data[0])
	

def set_conection_info(server,apikey):
	now = datetime.now().strftime("%m-%d-%Y %H:%M")
	cursor = conn.cursor()
	cursor.execute('UPDATE settings SET server = ?, apikey = ? WHERE id = 0',(server, apikey))
	logging.info(now + ' - Info : Connectioninformations updated')
	print('Your connection information has been updated.')
	conn.commit()


def copy_data():
	connection = get_settings('connection')
	if get_settings('copy_status') == 0:
		now = datetime.now().strftime("%m-%d-%Y %H:%M")
		cursor = conn.cursor()
		req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
		headers={"Content-Type": "application/json",
		'X-API-Key': connection['key']
		}
		)
		data = req.json()
		if not data:
			print('\n [b red]Error[/b red] : No aliases on server %s. Nothing to copy!\n\n' %(connection['server']))
			exit(0)
		i=0
		for data in data:
			cursor.execute('INSERT INTO aliases values(?,?,?,?)', (data['id'], data['address'],data['goto'],now))
			i=i+1
		cursor.execute('UPDATE settings SET data_copy = ? WHERE id = 0',(1,))
		conn.commit()
		logging.info(now + ' - Info : Imported %s new aliases from %s ' %(str(i),connection['server']))
		print('\n[b]Info[/b] : %s aliases imported from the mailcow instance %s to local DB\n' %(i, connection['server']))
	else:
		print('\n[b]Info[/b] : aliases alreday imported from the mailcow instance %s to local DB\n\n[i]Updating with any missing aliases![/i]' %(connection['server']))
		update_data()


def update_data():
	connection = get_settings('connection')
	if get_settings('copy_status') == 1:
		now = datetime.now().strftime("%m-%d-%Y %H:%M")
		req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
		headers={"Content-Type": "application/json",
		'X-API-Key': connection['key']
		}
		)
		data = req.json()
		i = 0
		count_alias = 0
		cursor = conn.cursor()
		for data in data:
			cursor.execute('SELECT count(*) FROM aliases where alias like ? and goto like ?', (data['address'],data['goto'],))
			count = cursor.fetchone()[0]
			if count >= 1 :
				i+=1
			else:
				cursor.execute('INSERT INTO aliases values(?,?,?,?)', (data['id'], data['address'],data['goto'],now))
				count_alias+=1
				i+=1
		conn.commit()
		if count_alias > 0:
			logging.info(now + ' - Info : Local DB updated with %s new aliases from %s ' %(str(count_alias),connection['server']))
			print('\n[b]Info[/b] : Local DB update with %s new aliases from %s \n' %(str(count_alias),connection['server']))
		else:
			print('\n[b]Info[/b] : No missing aliases from local DB \n')


def create(alias,to_address):
	now = datetime.now().strftime("%m-%d-%Y %H:%M")
	connection = get_settings('connection') # pyright: ignore
	check = checklist(alias)
	if check[0] == True:
		logging.error(now + ' - Error : alias %s exists on the mailcow instance %s ' %(alias,connection['server']))
		print('\n[b]Error[/b] : alias %s exists on the mailcow instance %s \n' %(alias,connection['server']))
		exit(0)
	elif check[1] == True:
		logging.error(now + ' - Error : alias %s exists in local database.' %(alias))
		print('\n[b]Error[/b] : alias %s exists in local database. \n' %(alias))
		exit(0)
	else:
		try:
			new_data = {'address': alias,'goto': to_address,'active': "1"}
			new_data = json.dumps(new_data)
			req = httpx.post('https://'+connection['server']+'/api/v1/add/alias',
			headers={"Content-Type": "application/json",
			'X-API-Key': connection['key']
			},
			data=new_data
			)
		except httpx.HTTPError as exc:
			print(f"Error while requesting {exc.request.url!r}.")
		mail_id = alias_id(alias)
		if mail_id == None:
			logging.error(now + ' - Error : alias %s not created on the mailcow instance %s ' %(alias,connection['server']))
			print('\n[b]Error[/b] : alias %s exists on the mailcow instance %s \n' %(alias,connection['server']))
		else:
			cursor = conn.cursor()
			cursor.execute('INSERT INTO aliases values(?,?,?,?)', (mail_id, alias,to_address,now))
			conn.commit()
			logging.info(now + ' - Info : alias %s created for %s on the mailcow instance %s ' %(alias,to_address,connection['server']))
			print('\n[b]Info[/b] : alias %s created for %s on the mailcow instance %s \n' %(alias,to_address,connection['server']))


def checklist(alias):
	alias_e = None
	alias_i = None
	connection = get_settings('connection')
	req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
		headers={"Content-Type": "application/json",
		'X-API-Key': connection['key']
		}
		)
	data = req.json()
	i = 0
	for data in data:
		if alias == data['address'] or alias in data['goto']:
			alias_e = True
		i=i+1
	cursor = conn.cursor()
	cursor.execute('SELECT count(*) FROM aliases where alias == ? or goto == ?', (alias,alias,))
	count = cursor.fetchone()[0]
	if count >= 1 :
		alias_i = True
	alias_exist = [alias_e,alias_i]
	return alias_exist


def alias_id(alias):
	connection = get_settings('connection')
	req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
		headers={"Content-Type": "application/json",
		'X-API-Key': connection['key']
		}
		)
	data = req.json()
	i = 0
	for data in data:
		if data['address'] == alias:
			return data['id']
		i=i+1
	return None


def number_of_aliases_in_db():
	cursor = conn.cursor()
	cursor.execute('SELECT count(*) FROM aliases')
	count = cursor.fetchone()[0]
	return count


def number_of_aliases_on_server():
	connection = get_settings('connection')
	req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
		headers={"Content-Type": "application/json",
		'X-API-Key': connection['key']
		}
		)
	data = req.json()
	return len(data)


def search(alias):
	connection = get_settings('connection')
	cursor = conn.cursor()
	search_term = '%'+alias+'%'
	cursor.execute('SELECT alias,goto from aliases where alias like ? or goto like ?',(search_term,search_term,))
	localdata = cursor.fetchall()
	req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
		headers={"Content-Type": "application/json",
		'X-API-Key': connection['key']
		}
		)
	remotedata = req.json()
	remote = []
	i=0
	for data in remotedata:
		if alias in remotedata[i]['address'] or alias in remotedata[i]['goto']:
			remote.append((remotedata[i]['address'], remotedata[i]['goto']))
		i=i+1
	finallist = localdata + list(set(remote) - set(localdata))
	i = 0
	print('\nAliases on %s containg search term [b]%s[/b]' %(connection['server'],alias))
	print('=================================================================')
	for data in finallist:
		the_alias = finallist[i][0].ljust(20,' ')
		print(the_alias + '\tgoes to\t\t' + finallist[i][1])
		i=i+1
	print('\n'+footer+'\n')


def get_mail_domains(info):
	connection = get_settings('connection')
	cursor = conn.cursor()
	req = httpx.get('https://'+connection['server']+'/api/v1/get/domain/all',
		headers={"Content-Type": "application/json",
		'X-API-Key': connection['key']
		}
		)
	remoteData = req.json()
	if info:
		total_aliases = 0
		i=0
		print('\n[b]malias[/b] - All email domains on %s' %(connection['server']))
		print('==================================================================')
		for domains in remoteData:
			cursor.execute('SELECT count(*) FROM aliases where alias like ? or goto like ?', ('%'+remoteData[i]['domain_name']+'%','%'+remoteData[i]['domain_name']+'%',))
			count = cursor.fetchone()[0]
			total_aliases += count
			print('%s \t\twith %s aliases' %(remoteData[i]['domain_name'],count))
			i+=1
		print('\n\nThere is a total of %s domains with %s aliases.\n%s' %(str(i),str(total_aliases),footer))

	else:
		return(remoteData)


def show_current_info():
	connection = get_settings('connection')
	latest_release = get_latest_release()
	aliases_server = number_of_aliases_on_server()
	alias_db = number_of_aliases_in_db()
	mail_domains = get_mail_domains(False)
	domain = ""
	i=0
	for domains in mail_domains:
		if i!=0:
			domain = domain + ', ' + str(mail_domains[i]['domain_name'])
		else:
			domain = domain + str(mail_domains[i]['domain_name'])
		i+=1
	print('\n[b]malias[/b] - Manage aliases on mailcow Instance.')
	print('===================================================')
	print('API key 		: [b]%s[/b]' % (connection['key']))
	print('Mailcow Instance 	: [b]%s[/b]' % (connection['server']))
	print('Active domains		: [b]%s[/b]' % (domain))
	print('Logfile 		: [b]%s[/b]' % (logfile))
	print('Databse 		: [b]%s[b]' % (database))
	print('Aliases on server 	: [b]%s[/b]' % (aliases_server))
	print('Aliases in DB 		: [b]%s[/b]' % (alias_db))
	print('')
	if float(app_version) < float(latest_release):
		print('App version 		: [b]%s[/b] a new version (%s) is available @ https://iurl.no/malias' % (app_version,latest_release))
	else:
		print('App version 		: [b]%s[/b]' % (app_version))
	print('')


def delete_alias(alias):
	status_e = None
	status_i = None
	now = datetime.now().strftime("%m-%d-%Y %H:%M")
	connection = get_settings('connection')
	check = checklist(alias)
	if check[0] == None and check[1] == None:
		print('\n[b]Error[/b] : The alias %s not found')
		exit(0)
	if check[0] or check[1] == True:
		alias_server_id = alias_id(alias)
		data = {'id': alias_server_id}
		delete_data = json.dumps(data)
		if check[0] == True and alias_server_id != None:
			req = httpx.post('https://'+connection['server']+'/api/v1/delete/alias',
				headers={"Content-Type": "application/json",
				'X-API-Key': connection['key']
				},
				data=delete_data
			)
		data=req.json()
		code = str(req)
		if code.find('200') != -1:
			status_e = True
		else:
			status_e = None
		if check[1] == True:
			cursor = conn.cursor()
			cursor.execute('DELETE from aliases where id = ?',(alias_server_id,))
			conn.commit()
			status_i = True
		if status_e == True and status_i == True:
			logging.info(now + ' - Info : alias %s deleted from the mailcow instance %s and Local DB' %(alias,connection['server']))
			print('\n[b]Info[/b] : alias %s deleted from the mailcow instance %s and local DB' %(alias,connection['server']))
		if status_e == True and status_i == None:
			logging.info(now + ' - Info : alias %s deleted from the mailcow instance %s.' %(alias,connection['server']))
			print('\n[b]Info[/b] : alias %s deleted from the mailcow instance %s.' %(alias,connection['server']))
		if status_e == None and status_i == True:
			logging.info(now + ' - Info : alias %s deleted from the local database.' %(alias))
			print('\n[b]Info[/b] : alias %s deleted from the local database.' %(alias))


def create_timed(username,domain):
	now = datetime.now().strftime("%m-%d-%Y %H:%M")
	connection = get_settings('connection')
	data = {'username': username,'domain': domain,'description': 'malias v'+app_version}
	data_json = json.dumps(data)
	req = httpx.post('https://'+connection['server']+'/api/v1/add/time_limited_alias',data=data_json,
				headers={"Content-Type": "application/json",
				'X-API-Key': connection['key']
				}
			)
	response = json.loads(req.text)
	if response[0]['type'] == 'danger' and response[0]['msg'] == 'domain_invalid':
		logging.error(now + ' - Error : the domain %s does not exist.' %(domain))
		print('[b][red]Error[/red][/b] : the domain %s does not exist.' %(domain))
		exit(0)
	if response[0]['type'] == 'danger' and response[0]['msg'] == 'access_denied':
		logging.error(now + ' - Error : something went wrong. The server responded with access denied.')
		print('[b][red]Error[/red][/b] : something went wrong. The server responded with [b]access denied[/b].')
		exit(0)
	alias = get_last_timed(username)
	validity = datetime.utcfromtimestamp(alias['validity']).strftime('%d-%m-%Y %H:%M')
	print('The timed alias %s was created. The alias is valid until %s UTC\n' %(alias['address'], validity))
	cursor = conn.cursor()
	cursor.execute('INSERT INTO timedaliases values(?,?,?,?)', (alias['validity'],alias['address'],username,validity))
	conn.commit()
	logging.info(now + ' - Info : timed alias %s created for %s and valid too %s UTC on the mailcow instance %s ' %(alias['address'],username,validity,connection['server']))


def check_local_db(alias_id):
	cursor = conn.cursor()
	cursor.execute('SELECT count(*) FROM aliases where id = ?',(alias_id,))
	count = cursor.fetchone()[0]
	return count


def list_alias():
	now = datetime.now().strftime("%m-%d-%Y %H:%M")
	connection = get_settings('connection')
	cursor = conn.cursor()
	req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
		headers={"Content-Type": "application/json",
		'X-API-Key': connection['key']
		}
		)
	data = req.json()
	i = 0
	l = 0
	print('\n[b]malias[/b] - All aliases on %s ([b]*[/b] also in local db)' %(connection['server']))
	print('==================================================================')
	for search in data:
		the_alias = data[i]['address'].ljust(20,' ')
		the_goto = data[i]['goto'].ljust(20,' ')
		cursor.execute('SELECT count(*) FROM aliases where alias like ? or goto like ?', (data[i]['address'],data[i]['address'],))
		count = cursor.fetchone()[0]
		if count >= 1:
			print(the_alias + '\tgoes to\t\t' + the_goto + '\t[b]*[/b]')
			l=l+1
		else:
			print(the_alias + '\tgoes to\t\t' + the_goto)
		i=i+1
	print('\n\nTotal number of aliases %s on instance [b]%s[/b] and %s on [b]local DB[/b].' %(str(i),connection['server'],str(l)))
	print('\n'+footer)


def export_data():
	connection = get_settings('connection')
	cursor = conn.cursor()
	req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
		headers={"Content-Type": "application/json",
		'X-API-Key': connection['key']
		}
		)
	data = req.json()
	with open("alias.json", "w") as outfile:
		json.dump(data, outfile, ensure_ascii=False, indent=4)



# For Testing purposes

def list_all():
	connection = get_settings('connection')
	req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
		headers={"Content-Type": "application/json",
		'X-API-Key': connection['key']
		}
		)
	data = req.json()
	print(data)


def updatedb():
	# Function for updatimg DB when we have to

	# 26.02.2025
	# Placeholder for future updates and functions.
	exit(1)


conn = connect_database()

parser = argparse.ArgumentParser(prog='malias',
								description='Malias is an application for adding, creating, and deleting aliases on a Mailcow instance. \n\nUse the issues section in the git repo for any problems or suggestions. https://gitlab.pm/rune/malias',
								formatter_class=RawTextHelpFormatter,
								epilog='Making mailcow easier...')
parser.add_argument('-c', '--copy', help='Copy alias data from mailcow server to local DB.\n\n',
					required=False, action='store_true')
parser.add_argument('-s', '--set', help='Set connection information.\n\n',
	nargs=2, metavar=('server', 'APIKey'), required=False, action="append")
parser.add_argument('-a', '--add', help='Add new alias.\n\n',
	nargs=2, metavar=('alias@domain.com', 'to@domain.com'), required=False, action="append")
parser.add_argument('-f', '--find', help='Search for alias.\n\n',
	nargs=1, metavar=('alias@domain.com'), required=False, action="append")
parser.add_argument('-d', '--delete', help='Delete alias.\n\n',
	nargs=1, metavar=('alias@domain.com'), required=False, action="append")
parser.add_argument('-t', '--timed', help='Add new time limited alias for user on domain. \nThe user@domain.com is where you want the alias to be delivered to.\nThe domain.com is which domain to use when creating the timed-alias.\nOne year validity\n\n',
	nargs=2, metavar=('user@domain.com', 'domain.com'), required=False, action="append")
parser.add_argument('-l', '--list', help='List all aliases on the Mailcow instance.\n\n',
					required=False, action='store_true')
parser.add_argument('-o', '--domains', help='List all mail domains on the Mailcow instance.\n\n',
					required=False, action='store_true')
parser.add_argument('-e', '--export', help='List all mail domains on the Mailcow instance.\n\n',
					required=False, action='store_true')
parser.add_argument('-v', '--version', help='Show current version and information\n\n',
					required=False, action='store_true')


args = vars(parser.parse_args())

if args['copy']:
	copy_data()
elif args['set']:
	set_conection_info(args['set'][0][0],args['set'][0][1])
elif args['add']:
	create(args['add'][0][0],args['add'][0][1])
elif args['find']:
	search(args['find'][0][0])
elif args['version']:
	show_current_info()
elif args['delete']:
	delete_alias(args['delete'][0][0])
elif args['timed']:
	create_timed(args['timed'][0][0],args['timed'][0][1])
elif args['list']:
	list_alias()
elif args['domains']:
	get_mail_domains(True)
elif args['export']:
	export_data()
else:
	print('\n\nEh, sorry! I need something more to help you! If you write [b]malias -h[/b] I\'ll show a help screen to get you going!!!\n\n\n')
	#export_data()