Added function to list timed aliases. Cleand up the code some

.gitignore

@@ -1,4 +1,6 @@
 list_alias.py
 malias.zip
 malias_local.py
+*.json
 .DS_Store
+.python-version

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,55 @@ For instructions run
 malias -h
 ```
 
-## Screenshot
+## Documentation
 
-![Screenshot of malias](https://gitlab.pm/rune/malias/raw/branch/main/malias.png)
+```bash
+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] [-w alias@domain.com] [-l] [-o] [-e] [-v]
+
+Malias is an application for adding, creating, and deleting aliases on a Mailcow instance.
+
+Use the issues section in the git repo for any problems or suggestions. https://gitlab.pm/rune/malias
+
+options:
+  -h, --help            show this help message and exit
+  -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
+
+  -w alias@domain.com, --alias alias@domain.com
+                        List timed (temprary) aliases connected toone account.
+
+  -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...
 
 ## Contributing
 

malias.py

@@ -1,4 +1,5 @@
-#!/usr/bin/python3
+#!/usr/bin/python3 -W ignore::DeprecationWarning
+import sys
 import sqlite3
 from pathlib import Path
 from sqlite3 import Error
@@ -6,39 +7,43 @@ import urllib.request
 import json
 import logging
 import argparse
-import requests
 import os
 import time
-import sys
-import git
-from types import SimpleNamespace
-from datetime import datetime
-from string import ascii_letters, digits
 from rich import print
+from datetime import datetime, timedelta
+from string import ascii_letters, digits
 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')
+filepath = homefilepath.joinpath('.config/malias2')
-database = filepath.joinpath('malias.db')
+database = filepath.joinpath('malias2.db')
-logfile = filepath.joinpath('malias.log')
+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.3.1'
+logging.getLogger("httpx").setLevel(logging.ERROR)
+app_version = '2.0'
+db_version = '2.0.0'
+footer = 'malias version %s'%(app_version)
+
+
+def unix_to_datetime(unix_timestamp):
+    return datetime.fromtimestamp(unix_timestamp)
 
 
 def get_latest_release():
-	req = urllib.request.Request('https://gitlab.pm/api/v1/repos/rune/malias/releases/latest')
+	version = 'N/A'
-	req.add_header('Content-Type', 'application/json')
+	req = httpx.get('https://gitlab.pm/api/v1/repos/rune/malias/releases/latest',
-	current = urllib.request.urlopen(req)
+		headers={"Content-Type": "application/json"}
-	remote = current.read().decode('utf-8')
+		)
-	remoteData = json.loads(remote)
+	version = req.json()['tag_name']
-	return remoteData['tag_name']
+	return version
 
 
 def release_check():
@@ -49,6 +54,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
@@ -60,13 +66,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
@@ -74,13 +78,23 @@ 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)''')
+			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()
@@ -88,270 +102,196 @@ 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 settings values(?,?,?,?,?)', (0, 1, 'dummy.server','DUMMY_KEY',0))
-		cursor.execute('INSERT INTO apikey values(?,?)', (1, 'DUMMY_KEY'))
+		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]
+	first_run_status = data[0][1] # pyright: ignore
-	mail_server = data[0][2]
+	server = data[0][2] # pyright: ignore
-	copy_status = data[0][3]
+	api_key = data[0][3] # pyright: ignore
-	if kind == 'mail_server':
+	copy_status = data[0][4] # pyright: ignore
-		if mail_server == 'dummy.server':
+	if kind == 'connection':
-			print('Error: No mailcow server active. Please add one with [b]malias -m [i]your.server[/i][/b]')
+		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):
-def get_api():
+	connection = get_settings('connection')
-	latest_release = get_latest_release()
+	req = httpx.get('https://'+connection['server']+'/api/v1/get/time_limited_aliases/%s' %username,
-	cursor = conn.cursor()
+		headers={"Content-Type": "application/json",
-	cursor.execute('SELECT api FROM apikey')
+		'X-API-Key': connection['key']
-	apikey = cursor.fetchone()[0]
+		}
-	if apikey == 'DUMMY_KEY':
+		)
-		print('Missing API key. Please add with [b]malias -k [i]YOUR-API-KEY[/i][/b]')
+	data = req.json()
-		exit(0)
+	data.sort(key = itemgetter('validity'), reverse=True)
-	else:
+	return(data[0])
-		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,))
+	cursor.execute('UPDATE settings SET server = ?, apikey = ? WHERE id = 0',(server, apikey))
-	logging.info(now + ' - Info : mailcow server updated')
+	logging.info(now + ' - Info : Connectioninformations updated')
-	print('Your mail server has been 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 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()
+	connection = get_settings('connection')
-	mail_server = get_settings('mail_server')
 	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 = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
-		req.add_header('Content-Type', 'application/json')
+		headers={"Content-Type": "application/json",
-		req.add_header('X-API-Key', apikey)
+		'X-API-Key': connection['key']
-		current = urllib.request.urlopen(req)
+		}
-		remote = current.read().decode('utf-8')
+		)
-		remoteData = json.loads(remote)
+		data = req.json()
-		i = 0
+		if not data:
-		for data in remoteData:
+			print('\n [b red]Error[/b red] : No aliases on server %s. Nothing to copy!\n\n' %(connection['server']))
-			cursor.execute('INSERT INTO aliases values(?,?,?,?)', (remoteData[i]['id'], remoteData[i]['address'],remoteData[i]['goto'],now))
+			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,))
-		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))
+		logging.info(now + ' - Info : Imported %s new aliases from %s ' %(str(i),connection['server']))
-		print('\n[b]Info[/b] : aliases imported from the mailcow instance %s to local DB\n' %(mail_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()
+	connection = get_settings('connection')
-	mail_server = get_settings('mail_server')
 	if get_settings('copy_status') == 1:
 		now = datetime.now().strftime("%m-%d-%Y %H:%M")
-		cursor = conn.cursor()
+		req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
-		req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all')
+		headers={"Content-Type": "application/json",
-		req.add_header('Content-Type', 'application/json')
+		'X-API-Key': connection['key']
-		req.add_header('X-API-Key', apikey)
+		}
-		current = urllib.request.urlopen(req)
+		)
-		remote = current.read().decode('utf-8')
+		data = req.json()
-		remoteData = json.loads(remote)
 		i = 0
 		count_alias = 0
 		cursor = conn.cursor()
-		for data in remoteData:
+		for data in data:
-			cursor.execute('SELECT count(*) FROM aliases where alias like ? and goto like ?', (remoteData[i]['address'],remoteData[i]['goto'],))
+			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))
+			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),mail_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
+	alias_e = None
-	apikey = get_api()
+	alias_i = None
-	mail_server = get_settings('mail_server')
+	connection = get_settings('connection')
-	req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all')
+	req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
-	req.add_header('Content-Type', 'application/json')
+		headers={"Content-Type": "application/json",
-	req.add_header('X-API-Key', apikey)
+		'X-API-Key': connection['key']
-	current = urllib.request.urlopen(req)
+		}
-	remote = current.read().decode('utf-8')
+		)
-	remoteData = json.loads(remote)
+	data = req.json()
 	i = 0
-	for search in remoteData:
+	for data in data:
-		if alias in remoteData[i]['address'] or alias in remoteData[i]['goto']:
+		if alias == data['address'] or alias in data['goto']:
-			alias_exist = True
+			alias_e = 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
+		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()
+	connection = get_settings('connection')
-	mail_server = get_settings('mail_server')
+	req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
-	req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all')
+		headers={"Content-Type": "application/json",
-	req.add_header('Content-Type', 'application/json')
+		'X-API-Key': connection['key']
-	req.add_header('X-API-Key', apikey)
+		}
-	current = urllib.request.urlopen(req)
+		)
-	remote = current.read().decode('utf-8')
+	data = req.json()
-	remoteData = json.loads(remote)
 	i = 0
-	for search in remoteData:
+	for data in data:
-		if remoteData[i]['address'] == alias:
+		if data['address'] == alias:
-			return remoteData[i]['id']
+			return data['id']
 		i=i+1
 	return None
 
@@ -363,103 +303,77 @@ def number_of_aliases_in_db():
 	return count
 
 
-
 def number_of_aliases_on_server():
-	apikey = get_api()
+	connection = get_settings('connection')
-	mail_server = get_settings('mail_server')
+	req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
-	req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all')
+		headers={"Content-Type": "application/json",
-	req.add_header('Content-Type', 'application/json')
+		'X-API-Key': connection['key']
-	req.add_header('X-API-Key', apikey)
+		}
-	current = urllib.request.urlopen(req)
+		)
-	remote = current.read().decode('utf-8')
+	data = req.json()
-	remoteData = json.loads(remote)
+	return len(data)
-	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()
+	connection = get_settings('connection')
-	mail_server = get_settings('mail_server')
-	alias_server = number_of_aliases_on_server()
-	alias_db = number_of_aliases_in_db()
 	cursor = conn.cursor()
-	cursor.execute('SELECT data_copy FROM settings where id = 1')
+	search_term = '%'+alias+'%'
-	result = cursor.fetchone()[0]
+	cursor.execute('SELECT alias,goto from aliases where alias like ? or goto like ?',(search_term,search_term,))
-	if result == 1:
+	localdata = cursor.fetchall()
-		search_term = '%'+alias+'%'
+	req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
-		cursor.execute('SELECT * from aliases where alias like ? or goto like ?',(search_term,search_term,))
+		headers={"Content-Type": "application/json",
-		remoteData = cursor.fetchall()
+		'X-API-Key': connection['key']
-		i = 0
+		}
-		print('\nAliases on %s that contains [b]%s[/b]' %(mail_server,alias))
+		)
-		print('=================================================================')
+	remotedata = req.json()
-		for search in remoteData:
+	remote = []
-			the_alias = remoteData[i][1].ljust(20,' ')
+	i=0
-			print(the_alias + '\tgoes to\t\t' + remoteData[i][2])
+	for data in remotedata:
-			i=i+1
+		if alias in remotedata[i]['address'] or alias in remotedata[i]['goto']:
-		print('\n\nData from local DB')
+			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')
+		return(remoteData)
-		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)
 
 
 def show_current_info():
-	API = get_api()
+	connection = get_settings('connection')
-	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()
 	mail_domains = get_mail_domains(False)
 	domain = ""
 	i=0
@@ -471,87 +385,236 @@ 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('API key 		: [b]%s[/b]' % (connection['key']))
-	print('Mailcow Instance 	: [b]%s[/b]' % (mail_server))
+	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 != 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 = unix_to_datetime(alias['validity'])
+	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)
+
+
+def list_timed_aliases(account):
+	connection = get_settings('connection')
+	req = httpx.get('https://'+connection['server']+'/api/v1/get/time_limited_aliases/'+account,
+		headers={"Content-Type": "application/json",
+		'X-API-Key': connection['key']
+		}
+		)
+	data = req.json()
+	i = 0
+	print('\n[b]malias[/b] - Timed aliases on %s for %s' %(connection['server'], account))
+	print('==========================================================================================================')
+	for data in data:
+		the_alias = data['address'].ljust(30,' ')
+		the_goto = data['goto'].ljust(20,' ')
+		the_validity = unix_to_datetime(data['validity'])
+		print(the_alias + '\tgoes to\t\t' + the_goto+'Valid to: '+str(the_validity))
+		i=i+1
+	#print('\n\nTotal number of timed aliases  on instance [b]%s[/b] for account.' %(connection['server'],account))
+	print('\n'+footer)
+
+
+# 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='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('-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('-v', '--version', help='Show current version\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('-w', '--alias', help='List timed (temprary) aliases connected toone account.\n\n',
+	nargs=1, metavar=('alias@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']:
+if args['copy']:
-	apikey(args['api'][0][0])
+	copy_data()
-elif args['search']:
+elif args['set']:
-	search(args['search'][0][0])
+	set_conection_info(args['set'][0][0],args['set'][0][1])
-elif args['server']:
-	set_mailserver(args['server'][0][0])
 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['info']:
+elif args['timed']:
-	show_current_info()
+	create_timed(args['timed'][0][0],args['timed'][0][1])
-elif args['version']:
+elif args['alias']:
-	release_check()
+	list_timed_aliases(args['alias'][0][0])
-elif args['copy']:
-	copy_data()
 elif args['list']:
 	list_alias()
 elif args['domains']:
 	get_mail_domains(True)
-		
+elif args['export']:
-	
+	export_data()
 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')
+	

requirements.txt

@@ -1,48 +1,13 @@
-aiohttp==3.8.4
+anyio==4.8.0
-aiosignal==1.3.1
+certifi==2025.1.31
-async-timeout==4.0.2
+exceptiongroup==1.2.2
-attrs==22.2.0
+h11==0.14.0
-bleach==6.0.0
+httpcore==1.0.7
-build==0.10.0
+httpx==0.28.1
-certifi==2022.12.7
+idna==3.10
-cffi==1.15.1
+markdown-it-py==3.0.0
-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
 mdurl==0.1.2
-more-itertools==9.1.0
+pygments==2.19.1
-multidict==6.0.4
+rich==13.9.4
-notmuch==0.37
+sniffio==1.3.1
-notmuch2==0.37
+typing-extensions==4.12.2
-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