Compare commits

..

12 Commits
0.2.3 ... main

4 changed files with 312 additions and 83 deletions

1
.gitignore vendored
View File

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

View File

@ -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

Binary file not shown.

Before

Width:  |  Height:  |  Size: 573 KiB

343
malias.py Executable file → Normal file
View File

@ -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,25 @@ 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.2.3'
app_version = '0.4'
db_version = '0.2'
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']
def release_check():
latest_release = get_latest_release()
if app_version != latest_release:
print('[b]New version available @ [i]https://iurl.no/malias[/b][/i]')
else:
print ('You have the the latest version. Version: %s' %(app_version))
def connect_database():
@ -57,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()
@ -76,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]
@ -95,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]
@ -107,28 +132,27 @@ 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()
def get_mail_domains():
def get_mail_domains(info):
apikey = get_api()
cursor = conn.cursor()
mail_server = get_settings('mail_server')
@ -138,20 +162,23 @@ def get_mail_domains():
current = urllib.request.urlopen(req)
remote = current.read().decode('utf-8')
remoteData = json.loads(remote)
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)))
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')
@ -164,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))
@ -175,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')
@ -186,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:
@ -208,6 +260,7 @@ def delete_alias(alias):
def copy_data():
apikey = get_api()
mail_server = get_settings('mail_server')
@ -224,14 +277,47 @@ 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))
print('\n[b]Info[/b] : aliases imported from the mailcow instance %s to local DB\n' %(mail_server))
else:
print('\n[b]Info[/b] : aliases alreday imported from the mailcow instance %s to local DB\n' %(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]' %(mail_server))
update_data()
def update_data():
apikey = get_api()
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 = 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
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'],))
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))
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))
else:
print('\n[b]Info[/b] : No missing aliases from local DB \n')
def checklist(alias):
alias_exist = None
@ -245,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
@ -283,8 +369,7 @@ def list_alias():
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()
@ -303,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')
@ -322,19 +420,20 @@ 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()
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')
result = cursor.fetchone()[0]
@ -343,13 +442,14 @@ def search(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 %s' %(mail_server,alias))
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')
else:
req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all')
req.add_header('Content-Type', 'application/json')
@ -358,13 +458,16 @@ def search(alias):
remote = current.read().decode('utf-8')
remoteData = json.loads(remote)
i = 0
print('\nAliases on %s that contains %s' %(mail_server,alias))
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():
@ -376,77 +479,154 @@ def get_mailcow_version():
current = urllib.request.urlopen(req)
remote = current.read().decode('utf-8')
remoteData = json.loads(remote)
#repo = Repo("https://github.com/mailcow/mailcow-dockerized")
# info = repo.git.ls_remote()
# tags = repo.tags
remote_heads = git.cmd.Git().ls_remote('https://github.com/mailcow/mailcow-dockerized', tags=True)
# tags = remote_heads.rstrip('refs/tags/')
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()
print('\n[b]malias[/b] - Manage aliases on mailcow Instance.')
mail_domains = get_mail_domains(False)
domain = ""
i=0
for domains in mail_domains:
if i!=0:
domain = domain + ', ' + str(mail_domains[i]['domain_name'])
else:
domain = domain + str(mail_domains[i]['domain_name'])
i+=1
print('\n[b]malias[/b] - Manage aliases on Mailcow instance.')
print('===================================================')
print('API key : [b]%s[/b]' % (API))
print('mailcow Instance : [b]%s[/b]' % (mail_server))
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("API key\t\t\t: [b]%s[/b]" % (API))
print("Mailcow Instance\t: [b]%s[/b]" % (mail_server))
print("Active domains\t\t: [b]%s[/b]" % (domain))
print("Mailcow version\t\t: [b]%s[/b]" % (mailcow_version))
print("Logfile\t\t\t: [b]%s[/b]" % (logfile))
print("Databse\t\t\t: [b]%s[b]" % (database))
print("Aliases on server\t: [b]%s[/b]" % (aliases_server))
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('')
print('App version : [b]%s[/b] (https://gitlab.pm/rune/malias)' % (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 version and config info\n\n',
# 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',
@ -469,18 +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']:
import_data(args['import'][0][0])
elif args['version']:
show_current_info()
elif args['copy']:
copy_data()
elif args['list']:
list_alias()
elif args['domains']:
get_mail_domains()
get_mail_domains(True)
else:
#get_mailcow_version()
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')