From 59cdf3c856a5333ea1425cf77e8819e42c9c05b7 Mon Sep 17 00:00:00 2001 From: Rune Olsen Date: Mon, 3 Mar 2025 13:57:02 +0100 Subject: [PATCH] Refactor and bumped to version 2.0 --- README.md | 62 ++-- malias.py | 824 +++++++++++++++++++++-------------------------- requirements.txt | 59 +--- 3 files changed, 415 insertions(+), 530 deletions(-) diff --git a/README.md b/README.md index 6797f2f..6c2bfc3 100644 --- a/README.md +++ b/README.md @@ -17,51 +17,49 @@ malias -h ## Documentation ```bash -usage: malias [-h] [-k APIkey] [-s alias@domain.com] [-m mailcow-server.tld] - [-a alias@domain.com to@domain.com] - [-t user@domain.com domain.com] [-d alias@domain.com] - [-i alias.json] [-v] [-c] [-l] [-o] +usage: malias [-h] [-c] [-s server APIKey] [-a alias@domain.com to@domain.com] [-f alias@domain.com] [-d alias@domain.com] [-t user@domain.com domain.com] [-l] [-o] [-e] [-v] -This is a simple application to help you create and delete aliases on a mailcow instance. -If you find any issues or would like to submit a PR - please head over to https://gitlab.pm/rune/malias. +Malias is an application for adding, creating, and deleting aliases on a Mailcow instance. -I hope this makes your mailcow life a bit easier! +Use the issues section in the git repo for any problems or suggestions. https://gitlab.pm/rune/malias -optional arguments: +options: -h, --help show this help message and exit - -k APIkey, --api APIkey - Add/Change API key. - - -s alias@domain.com, --search alias@domain.com - Search for alias. - - -m mailcow-server.tld, --server mailcow-server.tld - Add/Uppdate mailcow instance. - - -a alias@domain.com to@domain.com, --add alias@domain.com to@domain.com - Add new alias. - - -t user@domain.com domain.com, --timed user@domain.com domain.com - Add new time limited alias for user on domain. One year validity - - -d alias@domain.com, --delete alias@domain.com - Delete alias. - - -v, --version Show current version and information - -c, --copy Copy alias data from mailcow server to local DB. - + + -s, --set server APIKey + Set connection information. + + -a, --add alias@domain.com to@domain.com + Add new alias. + + -f, --find alias@domain.com + Search for alias. + + -d, --delete alias@domain.com + Delete alias. + + -t, --timed user@domain.com domain.com + Add new time limited alias for user on domain. + The user@domain.com is where you want the alias to be delivered to. + The domain.com is which domain to use when creating the timed-alias. + One year validity + -l, --list List all aliases on the Mailcow instance. - + -o, --domains List all mail domains on the Mailcow instance. - + + -e, --export List all mail domains on the Mailcow instance. + + -v, --version Show current version and information + Making mailcow easier... ``` ## Plans -I'm also working on functionality for exporting and importing aliases. As of version _0.4_ there is an export and an import fucntion that is not active in the app. Hopefully they will be finnished some day... +I'm also working on functionality for ~~exporting~~ and importing aliases. As of version _0.4_ there is an export and an import fucntion that is not active in the app. Hopefully they will be finnished some day... ## Contributing diff --git a/malias.py b/malias.py index fd49ce5..898cc56 100644 --- a/malias.py +++ b/malias.py @@ -7,43 +7,38 @@ import urllib.request import json import logging import argparse -try: - import requests -except ImportError: - sys.path.append("/opt/homebrew/lib/python3.11/site-packages") import os -import time -import git -from types import SimpleNamespace +import time +from rich import print from datetime import datetime from string import ascii_letters, digits -from rich import print from argparse import RawTextHelpFormatter -from urllib.request import urlopen 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/malias') -database = filepath.joinpath('malias.db') -logfile = filepath.joinpath('malias.log') +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') -app_version = '0.4' -db_version = '0.2' - +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(): - req = urllib.request.Request('https://gitlab.pm/api/v1/repos/rune/malias/releases/latest') - req.add_header('Content-Type', 'application/json') - current = urllib.request.urlopen(req) - remote = current.read().decode('utf-8') - remoteData = json.loads(remote) - return remoteData['tag_name'] + 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(): @@ -54,6 +49,7 @@ def release_check(): print ('You have the the latest version. Version: %s' %(app_version)) + def connect_database(): Path(filepath).mkdir(parents=True, exist_ok=True) conn = None @@ -65,13 +61,11 @@ def connect_database(): finally: if conn: c = conn.cursor() - c.execute('''CREATE TABLE IF NOT EXISTS apikey - (id integer NOT NULL PRIMARY KEY, - api text NOT NULL)''') 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 @@ -84,13 +78,18 @@ def connect_database(): 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() @@ -98,312 +97,200 @@ def first_run(conn): count = cursor.fetchone()[0] if count == 0: logging.error(now + ' - First run!') - cursor.execute('INSERT INTO settings values(?,?,?,?)', (1, 1, 'dummy.server',0)) - cursor.execute('INSERT INTO apikey values(?,?)', (1, 'DUMMY_KEY')) + 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] - mail_server = data[0][2] - copy_status = data[0][3] - if kind == 'mail_server': - if mail_server == 'dummy.server': - print('Error: No mailcow server active. Please add one with [b]malias -m [i]your.server[/i][/b]') + 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: - return mail_server + 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 get_api(): - latest_release = get_latest_release() - cursor = conn.cursor() - cursor.execute('SELECT api FROM apikey') - apikey = cursor.fetchone()[0] - if apikey == 'DUMMY_KEY': - print('Missing API key. Please add with [b]malias -k [i]YOUR-API-KEY[/i][/b]') - exit(0) - else: - return apikey - - -def set_mailserver(server): +def set_conection_info(server,apikey): now = datetime.now().strftime("%m-%d-%Y %H:%M") cursor = conn.cursor() - cursor.execute('UPDATE settings SET server = ? WHERE id = 1',(server,)) - logging.info(now + ' - Info : mailcow server updated') - print('Your mail server has been updated.') + 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 apikey(key): - now = datetime.now().strftime("%m-%d-%Y %H:%M") - cursor = conn.cursor() - cursor.execute('UPDATE apikey SET api = ? WHERE id = 1',(key,)) - logging.info(now + ' - Info : API key updated') - print('Your API key has been updated.') - conn.commit() - - -def get_mail_domains(info): - apikey = get_api() - cursor = conn.cursor() - mail_server = get_settings('mail_server') - req = urllib.request.Request('https://'+mail_server+'/api/v1/get/domain/all') - req.add_header('Content-Type', 'application/json') - req.add_header('X-API-Key', apikey) - current = urllib.request.urlopen(req) - remote = current.read().decode('utf-8') - remoteData = json.loads(remote) - if info: - total_aliases = 0 - i=0 - print('\n[b]malias[/b] - All email domains on %s' %(mail_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.' %(str(i),str(total_aliases))) - - else: - return(remoteData) - - -def create(alias,to_address): - now = datetime.now().strftime("%m-%d-%Y %H:%M") - server = get_settings('mail_server') - apikey = get_api() - if checklist(alias) == True: - logging.error(now + ' - Error : alias %s exists on the mailcow instance %s ' %(alias,server)) - print('\n[b]Error[/b] : alias %s exists on the mailcow instance %s \n' %(alias,server)) - exit(0) - else: - data = {'address': alias,'goto': to_address,'active': "1"} - headers = {'X-API-Key': apikey, "Content-Type": "application/json"} - response = requests.post('https://'+server+'/api/v1/add/alias', - data=json.dumps(data), headers=headers) - mail_id = alias_id(alias) - if mail_id == None: - logging.error(now + ' - Error : alias %s not created on the mailcow instance %s ' %(alias,server)) - print('\n[b]Error[/b] : alias %s exists on the mailcow instance %s \n' %(alias,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,server)) - print('\n[b]Info[/b] : alias %s created for %s on the mailcow instance %s \n' %(alias,to_address,server)) - - -def create_timed(username,domain): - now = datetime.now().strftime("%m-%d-%Y %H:%M") - server = get_settings('mail_server') - apikey = get_api() - data = {'username': username,'domain': domain} - headers = {'X-API-Key': apikey, "Content-Type": "application/json"} - response = requests.post('https://'+server+'/api/v1/add/time_limited_alias', - data=json.dumps(data), headers=headers) - - response_data = response.json() - if response_data[0]['type'] == 'danger': - if response_data[0]['msg'] == 'domain_invalid': - print('\n[b]Error[/b] : the domain %s is invalid for %s \n' %(domain, server)) - if response_data[0]['msg'] == 'access_denied': - print('\n[b]Error[/b] : the server %s responded with [red]Access Denied[/red]\n' %(server)) - else: - 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,server)) - - - -def delete_alias(alias): - server = get_settings('mail_server') - apikey = get_api() - if checklist(alias) == True: - the_alias_id = alias_id(alias) - data = {'id': the_alias_id} - headers = {'X-API-Key': apikey, "Content-Type": "application/json"} - response = requests.post('https://'+server+'/api/v1/delete/alias', - data=json.dumps(data), headers=headers) - response_data = response.json() - if response_data[0]['type'] == 'success': - if check_local_db(the_alias_id) == 1: - now = datetime.now().strftime("%m-%d-%Y %H:%M") - cursor = conn.cursor() - cursor.execute('DELETE from aliases where id = ?',(the_alias_id,)) - logging.info(now + ' - Info : alias %s deleted from the mailcow instance %s and Local DB' %(alias,server)) - conn.commit() - print('\n[b]Info[/b] : alias %s deleted from the mailcow instance %s and local DB' %(alias,server)) - else: - logging.info(now + ' - Info : alias %s deleted from the mailcow instance %s.' %(alias,server)) - print('\n[b]Info[/b] : alias %s deleted from the mailcow instance %s.' %(alias,server)) - else: - logging.error(now + ' - Error : alias %s NOT deleted from the mailcow instance %s. Error : %s' %(alias,server,response_data)) - conn.commit() - print('\n[b]Error[/b] : alias %s NOT deleted from the mailcow instance %s. Error : %s' %(alias,server,response_data)) - else: - print('\n[b]Error[/b] : The alias %s not found') - - - - def copy_data(): - apikey = get_api() - mail_server = get_settings('mail_server') + connection = get_settings('connection') if get_settings('copy_status') == 0: now = datetime.now().strftime("%m-%d-%Y %H:%M") cursor = conn.cursor() - req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all') - req.add_header('Content-Type', 'application/json') - req.add_header('X-API-Key', apikey) - current = urllib.request.urlopen(req) - remote = current.read().decode('utf-8') - remoteData = json.loads(remote) - i = 0 - for data in remoteData: - cursor.execute('INSERT INTO aliases values(?,?,?,?)', (remoteData[i]['id'], remoteData[i]['address'],remoteData[i]['goto'],now)) + 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 = 1',(1,)) + cursor.execute('UPDATE settings SET data_copy = ? WHERE id = 0',(1,)) conn.commit() - logging.info(now + ' - Info : aliases imported from the mailcow instance %s to local DB' %(mail_server)) - print('\n[b]Info[/b] : aliases imported from the mailcow instance %s to local DB\n' %(mail_server)) + 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]' %(mail_server)) + 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(): - apikey = get_api() - mail_server = get_settings('mail_server') + connection = get_settings('connection') if get_settings('copy_status') == 1: now = datetime.now().strftime("%m-%d-%Y %H:%M") - cursor = conn.cursor() - req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all') - req.add_header('Content-Type', 'application/json') - req.add_header('X-API-Key', apikey) - current = urllib.request.urlopen(req) - remote = current.read().decode('utf-8') - remoteData = json.loads(remote) + 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 remoteData: - cursor.execute('SELECT count(*) FROM aliases where alias like ? and goto like ?', (remoteData[i]['address'],remoteData[i]['goto'],)) + 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(?,?,?,?)', (remoteData[i]['id'], remoteData[i]['address'],remoteData[i]['goto'],now)) + 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),mail_server)) - print('\n[b]Info[/b] : Local DB update with %s new aliases from %s \n' %(str(count_alias),mail_server)) + 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_exist = None - apikey = get_api() - mail_server = get_settings('mail_server') - req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all') - req.add_header('Content-Type', 'application/json') - req.add_header('X-API-Key', apikey) - current = urllib.request.urlopen(req) - remote = current.read().decode('utf-8') - remoteData = json.loads(remote) + 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 search in remoteData: - if alias == remoteData[i]['address'] == alias in remoteData[i]['goto']: - alias_exist = True + 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_exist = True - + alias_i = True + alias_exist = [alias_e,alias_i] return alias_exist -def list_alias(): - apikey = get_api() - cursor = conn.cursor() - mail_server = get_settings('mail_server') - req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all') - req.add_header('Content-Type', 'application/json') - req.add_header('X-API-Key', apikey) - current = urllib.request.urlopen(req) - remote = current.read().decode('utf-8') - remoteData = json.loads(remote) - i = 0 - l = 0 - print('\n[b]malias[/b] - All aliases on %s ([b]*[/b] also in local db)' %(mail_server)) - print('==================================================================') - for search in remoteData: - the_alias = remoteData[i]['address'].ljust(20,' ') - the_goto = remoteData[i]['goto'].ljust(20,' ') - cursor.execute('SELECT count(*) FROM aliases where alias like ? or goto like ?', (remoteData[i]['address'],remoteData[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),mail_server,str(l))) - - def alias_id(alias): - apikey = get_api() - mail_server = get_settings('mail_server') - req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all') - req.add_header('Content-Type', 'application/json') - req.add_header('X-API-Key', apikey) - current = urllib.request.urlopen(req) - remote = current.read().decode('utf-8') - remoteData = json.loads(remote) + 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 search in remoteData: - if remoteData[i]['address'] == alias: - return remoteData[i]['id'] + for data in data: + if data['address'] == alias: + return data['id'] i=i+1 return None -def get_last_timed(username): - apikey = get_api() - mail_server = get_settings('mail_server') - req = urllib.request.Request('https://'+mail_server+'/api/v1/get/time_limited_aliases/%s' %username) - req.add_header('Content-Type', 'application/json') - req.add_header('X-API-Key', apikey) - current = urllib.request.urlopen(req) - remote = current.read().decode('utf-8') - remoteData = json.loads(remote) - remoteData.sort(key = itemgetter('validity'), reverse=True) - return (remoteData[0]) - - def number_of_aliases_in_db(): cursor = conn.cursor() cursor.execute('SELECT count(*) FROM aliases') @@ -411,103 +298,77 @@ def number_of_aliases_in_db(): return count - def number_of_aliases_on_server(): - apikey = get_api() - mail_server = get_settings('mail_server') - req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all') - req.add_header('Content-Type', 'application/json') - req.add_header('X-API-Key', apikey) - current = urllib.request.urlopen(req) - remote = current.read().decode('utf-8') - remoteData = json.loads(remote) - return len(remoteData) - - -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 - + 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): - apikey = get_api() - mail_server = get_settings('mail_server') - alias_server = number_of_aliases_on_server() - alias_db = number_of_aliases_in_db() + connection = get_settings('connection') cursor = conn.cursor() - cursor.execute('SELECT data_copy FROM settings where id = 1') - result = cursor.fetchone()[0] - if result == 1: - search_term = '%'+alias+'%' - cursor.execute('SELECT * from aliases where alias like ? or goto like ?',(search_term,search_term,)) - remoteData = cursor.fetchall() - i = 0 - print('\nAliases on %s that contains [b]%s[/b]' %(mail_server,alias)) - print('=================================================================') - for search in remoteData: - the_alias = remoteData[i][1].ljust(20,' ') - print(the_alias + '\tgoes to\t\t' + remoteData[i][2]) - i=i+1 - print('\n\nData from local DB') + 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: - req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all') - req.add_header('Content-Type', 'application/json') - req.add_header('X-API-Key', apikey) - current = urllib.request.urlopen(req) - remote = current.read().decode('utf-8') - remoteData = json.loads(remote) - i = 0 - print('\nAliases on %s that contains [b]%s[/b]' %(mail_server,alias)) - print('=================================================') - for search in remoteData: - if alias in remoteData[i]['address'] or alias in remoteData[i]['goto']: - print(remoteData[i]['address'] + '\tgoes to\t\t' + remoteData[i]['goto']) - i=i+1 - print('\n\nData from server') - if alias_server != alias_db: - print('\n\nThere are %s aliases on the server and %s aliases in local DB' %(str(alias_server),str(alias_db))) - print('Run [i]malias -c[/i] to update local DB') - - -def get_mailcow_version(): - apikey = get_api() - mail_server = get_settings('mail_server') - req = urllib.request.Request('https://'+mail_server+'/api/v1/get/status/version') - req.add_header('Content-Type', 'application/json') - req.add_header('X-API-Key', apikey) - current = urllib.request.urlopen(req) - remote = current.read().decode('utf-8') - remoteData = json.loads(remote) - remote_heads = git.cmd.Git().ls_remote('https://github.com/mailcow/mailcow-dockerized', tags=True) - tags = remote_heads.splitlines() - for x in tags: - string = x.rsplit('/',2) - - if remoteData['version'] != string[2]: - versionInfo = 'Your Mailcow version is %s and the latest is %s' %(remoteData['version'], string[2]) - else: - versionInfo = 'You have the latest Mailcow version %s' %remoteData['version'] - - return (versionInfo) + return(remoteData) def show_current_info(): - API = get_api() - mail_server = get_settings('mail_server') + connection = get_settings('connection') latest_release = get_latest_release() - if API == 'DUMMY_KEY': - API = 'Missing API Key!' - - if mail_server == 'dummy.server': - mail_server = 'Missing address to mailcow instance!' - aliases_server = number_of_aliases_on_server() alias_db = number_of_aliases_in_db() - mailcow_version = get_mailcow_version() mail_domains = get_mail_domains(False) domain = "" i=0 @@ -519,151 +380,212 @@ def show_current_info(): i+=1 print('\n[b]malias[/b] - Manage aliases on mailcow Instance.') print('===================================================') - print('API key : [b]%s[/b]' % (API)) - print('Mailcow Instance : [b]%s[/b]' % (mail_server)) + print('API key : [b]%s[/b]' % (connection['key'])) + print('Mailcow Instance : [b]%s[/b]' % (connection['server'])) print('Active domains : [b]%s[/b]' % (domain)) - print('Mailcow version : [b]%s[/b]' % (mailcow_version)) 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 app_version[:3] != latest_release: + 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(): - apikey = get_api() - mail_server = get_settings('mail_server') - req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all') - req.add_header('Content-Type', 'application/json') - req.add_header('X-API-Key', apikey) - current = urllib.request.urlopen(req) - remote = current.read().decode('utf-8') - remoteData = json.dumps(json.loads(remote),indent=4) + 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: - outfile.write(remoteData) + json.dump(data, outfile, ensure_ascii=False, indent=4) -def import_data(file): - apikey=get_api() - mailserver = get_settings('mail_server') - active_domains = get_mail_domains(False) - with open(file,'r') as mail_alias: - json_object = json.load(mail_alias) - domain_list = [] - active_domain_list = [] - for data in json_object: - domain = data['domain'] - if domain not in domain_list: - domain_list.append(domain) - for data in active_domains: - active_domain_list.append(data['domain_name']) - diff = list(set(domain_list) - set(active_domain_list)) - if len(diff) != 0: - missing = ', '.join(diff) - print ('Please add the domains %s to your mailcow instance' % (missing)) - sys.exit(1) - else: - print('OK') +# 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 - # 09.04.24 - # Added DB tempalias table - # Added DB version table - cursor = conn.cursor() - cursor.execute('''CREATE TABLE IF NOT EXISTS dbversion - (version integer NOT NULL DEFAULT 0)''') - cursor.execute('SELECT COUNT(version) FROM dbversion') - count = cursor.fetchone()[0] - if count == 0: - cursor.execute('INSERT INTO dbversion values(?)', (db_version,)) - - conn.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() - + # 26.02.2025 + # Placeholder for future updates and functions. + exit(1) conn = connect_database() -updatedb() # Check if db needs updating and do the updating. parser = argparse.ArgumentParser(prog='malias', - description='This is a simple application to help you create and delete aliases on a mailcow instance.\nIf you find any issues or would like to submit a PR - please head over to https://gitlab.pm/rune/malias. \n\nI hope this makes your mailcow life a bit easier!', + 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('-k', '--api', help='Add/Change API key.\n\n', - nargs=1, metavar=('APIkey'), required=False, action="append") - -parser.add_argument('-s', '--search', help='Search for alias.\n\n', - nargs=1, metavar=('alias@domain.com'), required=False, action="append") - -parser.add_argument('-m', '--server', help='Add/Uppdate mailcow instance.\n\n', - nargs=1, metavar=('mailcow-server.tld'), 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('-t', '--timed', help='Add new time limited alias for user on domain. One year validity\n\n', - nargs=2, metavar=('user@domain.com', '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('-i', '--import', help='Show current config and appliacation info\n\n', -# nargs=1, metavar=('alias.json'), required=False, action="append") - -parser.add_argument('-v', '--version', help='Show current version and information\n\n', - required=False, action='store_true') - 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['api']: - apikey(args['api'][0][0]) -elif args['search']: - search(args['search'][0][0]) -elif args['server']: - set_mailserver(args['server'][0][0]) +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['timed']: - create_timed(args['timed'][0][0],args['timed'][0][1]) -elif args['delete']: - delete_alias(args['delete'][0][0]) -#elif args['import']: -# import_data(args['import'][0][0]) +elif args['find']: + search(args['find'][0][0]) elif args['version']: show_current_info() -elif args['copy']: - copy_data() +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() diff --git a/requirements.txt b/requirements.txt index 6f6610a..79801c6 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,48 +1,13 @@ -aiohttp==3.8.4 -aiosignal==1.3.1 -async-timeout==4.0.2 -attrs==22.2.0 -bleach==6.0.0 -build==0.10.0 -certifi==2022.12.7 -cffi==1.15.1 -charset-normalizer==3.1.0 -click==8.1.3 -docopt==0.6.2 -docutils==0.19 -frozenlist==1.3.3 -gitdb==4.0.10 -GitPython==3.1.32 -gpg==1.21.0 -idna==3.4 -importlib-metadata==6.1.0 -jaraco.classes==3.2.3 -keyring==23.13.1 -markdown-it-py==2.2.0 +anyio==4.8.0 +certifi==2025.1.31 +exceptiongroup==1.2.2 +h11==0.14.0 +httpcore==1.0.7 +httpx==0.28.1 +idna==3.10 +markdown-it-py==3.0.0 mdurl==0.1.2 -more-itertools==9.1.0 -multidict==6.0.4 -notmuch==0.37 -notmuch2==0.37 -openai==0.27.0 -packaging==23.0 -pipreqs==0.4.11 -pkginfo==1.9.6 -promptcli==1.0.4 -pycparser==2.21 -Pygments==2.14.0 -pyproject_hooks==1.0.0 -readme-renderer==37.3 -requests==2.28.2 -requests-toolbelt==0.10.1 -rfc3986==2.0.0 -rich==13.3.1 -six==1.16.0 -smmap==5.0.0 -tqdm==4.65.0 -twine==4.0.2 -urllib3==1.26.14 -webencodings==0.5.1 -yarg==0.1.9 -yarl==1.8.2 -zipp==3.15.0 +pygments==2.19.1 +rich==13.9.4 +sniffio==1.3.1 +typing-extensions==4.12.2