From 9cc931c5852cbead4642fdc2e915337d82812cb0 Mon Sep 17 00:00:00 2001 From: rune Date: Wed, 10 Apr 2024 13:40:00 +0200 Subject: [PATCH] Added timed aliases and edited README.md --- .gitignore | 1 + README.md | 52 ++++++++++++- malias.py | 212 ++++++++++++++++++++++++++++++++++++++++------------- 3 files changed, 212 insertions(+), 53 deletions(-) mode change 100755 => 100644 malias.py diff --git a/.gitignore b/.gitignore index d829215..835adad 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ list_alias.py malias.zip malias_local.py +*.json .DS_Store \ No newline at end of file diff --git a/README.md b/README.md index 161f69c..fd8c3e4 100644 --- a/README.md +++ b/README.md @@ -14,9 +14,57 @@ For instructions run malias -h ``` -## Screenshot +## Documentation -![Screenshot of malias](https://gitlab.pm/rune/malias/raw/branch/main/malias.png) +```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] + +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. + +I hope this makes your mailcow life a bit easier! + +optional arguments: + -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. + + -i alias.json, --import alias.json + Show current config and appliacation info + + -v, --version Show current version and information + + -c, --copy Copy alias data from mailcow server to local DB. + + -l, --list List all aliases on the Mailcow instance. + + -o, --domains List all mail domains on the Mailcow instance. + + +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... ## Contributing diff --git a/malias.py b/malias.py old mode 100755 new mode 100644 index 711a3f4..869bc0f --- a/malias.py +++ b/malias.py @@ -17,6 +17,7 @@ from string import ascii_letters, digits from rich import print from argparse import RawTextHelpFormatter from urllib.request import urlopen +from operator import itemgetter # Info pages for dev # https://mailcow.docs.apiary.io/#reference/aliases/get-aliases/get-aliases @@ -29,7 +30,8 @@ database = filepath.joinpath('malias.db') logfile = filepath.joinpath('malias.log') Path(filepath).mkdir(parents=True, exist_ok=True) logging.basicConfig(filename=logfile,level=logging.INFO,format='%(message)s') -app_version = '0.3.2' +app_version = '0.4' +db_version = '0.2' def get_latest_release(): @@ -47,7 +49,7 @@ def release_check(): 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) @@ -74,13 +76,18 @@ def connect_database(): 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)''') + conn.commit() first_run(conn) return conn - - + + def first_run(conn): now = datetime.now().strftime("%m-%d-%Y %H:%M") cursor = conn.cursor() @@ -93,12 +100,12 @@ def first_run(conn): conn.commit() return None - - + + def get_settings(kind): cursor = conn.cursor() cursor.execute('SELECT * FROM settings') - data = cursor.fetchall() + data = cursor.fetchall() first_run_status = data[0][1] mail_server = data[0][2] copy_status = data[0][3] @@ -112,10 +119,11 @@ def get_settings(kind): return first_run_status if kind == 'copy_status': return copy_status - - + + def get_api(): + latest_release = get_latest_release() cursor = conn.cursor() cursor.execute('SELECT api FROM apikey') apikey = cursor.fetchone()[0] @@ -124,23 +132,23 @@ def get_api(): exit(0) else: return apikey - + def set_mailserver(server): 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.') + print('Your mail server 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.') + print('Your API key has been updated.') conn.commit() @@ -170,7 +178,7 @@ def get_mail_domains(info): else: return(remoteData) - + def create(alias,to_address): now = datetime.now().strftime("%m-%d-%Y %H:%M") server = get_settings('mail_server') @@ -183,7 +191,7 @@ def create(alias,to_address): 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) + 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)) @@ -194,8 +202,33 @@ def create(alias,to_address): 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') @@ -205,7 +238,7 @@ def delete_alias(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) + data=json.dumps(data), headers=headers) response_data = response.json() if response_data[0]['type'] == 'success': if check_local_db(the_alias_id) == 1: @@ -244,7 +277,7 @@ def copy_data(): for data in remoteData: cursor.execute('INSERT INTO aliases values(?,?,?,?)', (remoteData[i]['id'], remoteData[i]['address'],remoteData[i]['goto'],now)) i=i+1 - + cursor.execute('UPDATE settings SET data_copy = ? WHERE id = 1',(1,)) conn.commit() logging.info(now + ' - Info : aliases imported from the mailcow instance %s to local DB' %(mail_server)) @@ -298,7 +331,7 @@ def checklist(alias): remoteData = json.loads(remote) i = 0 for search in remoteData: - if alias == remoteData[i]['address'] or alias == remoteData[i]['goto']: + if alias == remoteData[i]['address'] == alias in remoteData[i]['goto']: alias_exist = True i=i+1 cursor = conn.cursor() @@ -306,7 +339,7 @@ def checklist(alias): count = cursor.fetchone()[0] if count >= 1 : alias_exist = True - + return alias_exist @@ -355,6 +388,19 @@ def alias_id(alias): 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') @@ -374,14 +420,14 @@ def number_of_aliases_on_server(): 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 - + def search(alias): apikey = get_api() @@ -423,7 +469,7 @@ def search(alias): 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') @@ -437,25 +483,25 @@ def get_mailcow_version(): 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) - - + + def show_current_info(): API = get_api() - mail_server = get_settings('mail_server') + mail_server = get_settings('mail_server') 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() @@ -479,42 +525,105 @@ def show_current_info(): print('Aliases on server : [b]%s[/b]' % (aliases_server)) print('Aliases in DB : [b]%s[/b]' % (alias_db)) print('') - if app_version != latest_release: + if app_version[:5] != 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('') - - -conn = connect_database() +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) + with open("alias.json", "w") as outfile: + outfile.write(remoteData) + + +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') + +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() + + + +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!', formatter_class=RawTextHelpFormatter, epilog='Making mailcow easier...') -parser.add_argument('-k', '--api', help='Add/Change API key.\n\n', +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', +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', +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', +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('-d', '--delete', help='Delete alias.\n\n', +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', '--info', help='Show current config and appliacation info\n\n', - required=False, action='store_true') +# 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\n\n', +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', @@ -537,20 +646,21 @@ elif args['server']: set_mailserver(args['server'][0][0]) 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['info']: - show_current_info() +elif args['import']: + import_data(args['import'][0][0]) elif args['version']: - release_check() + show_current_info() elif args['copy']: copy_data() elif args['list']: list_alias() elif args['domains']: get_mail_domains(True) - - + + else: - # get_api() - 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') \ No newline at end of file + 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')