Work In Progress (WIP)

.gitignore

@@ -1,4 +1,5 @@
 list_alias.py
 malias.zip
 malias_local.py
+*.json
 .DS_Store

README.md

@@ -4,7 +4,7 @@ _malias_ is a helper for mailcow instances. You can create, delete, search and l
 
 ## Installation
 
-Download the latest release from https://gitlab.pm/rune/malias/releases. Unzip and move to a folder in you path (ease of use). I also recommend rename the file ```malias.py``` to just ```malias``` and make the file executable with ```chmod +x malias```. To install required python modules run ```pip3 install -r requirements.txt```
+Download the latest release from [https://gitlab.pm/rune/malias/releases](https://iurl.no/release). Unzip and move to a folder in you path (ease of use). I also recommend rename the file ```malias.py``` to just ```malias``` and make the file executable with ```chmod +x malias```. To install required python modules run ```pip3 install -r requirements.txt```
 
 ## Usage
 
@@ -14,9 +14,54 @@ 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.
+                        
+  -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
 

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.1'
+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,9 +119,9 @@ 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()
@@ -125,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()
 
 
@@ -171,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')
@@ -184,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))
@@ -195,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')
@@ -206,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:
@@ -245,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))
@@ -299,15 +331,15 @@ def checklist(alias):
 	remoteData = json.loads(remote)
 	i = 0
 	for search in remoteData:
-		if alias in remoteData[i]['address'] or alias in remoteData[i]['goto']:
+		if alias == remoteData[i]['address'] || alias in remoteData[i]['goto']:
 			alias_exist = True
 		i=i+1
 	cursor = conn.cursor()
-	cursor.execute('SELECT count(*) FROM aliases where alias like ? or goto like ?', (alias,alias,))
+	cursor.execute('SELECT count(*) FROM aliases where alias == ? or goto == ?', (alias,alias,))
 	count = cursor.fetchone()[0]
 	if count >= 1 :
 		alias_exist = True
-		
+
 	return alias_exist
 
 
@@ -356,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')
@@ -375,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()
@@ -424,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')
@@ -438,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()
@@ -469,53 +514,119 @@ def show_current_info():
 		else:
 			domain = domain + str(mail_domains[i]['domain_name'])
 		i+=1
-	print('\n[b]malias[/b] - Manage aliases on mailcow Instance.')
+	print('\n[b]malias[/b] - Manage aliases on Mailcow instance.')
 	print('===================================================')
-	print('API key 		: [b]%s[/b]' % (API))
+	print("API key\t\t\t: [b]%s[/b]" % (API))
-	print('Mailcow Instance 	: [b]%s[/b]' % (mail_server))
+    print("Mailcow Instance\t: [b]%s[/b]" % (mail_server))
-	print('Active domains		: [b]%s[/b]' % (domain))
+    print("Active domains\t\t: [b]%s[/b]" % (domain))
-	print('Mailcow version 	: [b]%s[/b]' % (mailcow_version))
+    print("Mailcow version\t\t: [b]%s[/b]" % (mailcow_version))
-	print('Logfile 		: [b]%s[/b]' % (logfile))
+    print("Logfile\t\t\t: [b]%s[/b]" % (logfile))
-	print('Databse 		: [b]%s[b]' % (database))
+    print("Databse\t\t\t: [b]%s[b]" % (database))
-	print('Aliases on server 	: [b]%s[/b]' % (aliases_server))
+    print("Aliases on server\t: [b]%s[/b]" % (aliases_server))
-	print('Aliases in DB 		: [b]%s[/b]' % (alias_db))
+    print("Aliases in DB\t\t: [b]%s[/b]" % (alias_db))
+    print("")
+    if app_version[:3] != latest_release:
+        print(
+            "App version\t\t\t\t: [b]%s[/b] a new version (%s) is available @ https://iurl.no/malias"
+            % (app_version, latest_release)
+        )
+    else:
+        print("App version\t\t: [b]%s[/b]" % (app_version))
 	print('')
-	if app_version != 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',
+# parser.add_argument('-i', '--import', help='Show current config and appliacation info\n\n',
-					required=False, action='store_true')
+# 	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',
@@ -538,20 +649,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']:
+elif args['import']:
-	show_current_info()
+	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')
-	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')