Compare commits

..

No commits in common. "main" and "0.3" have entirely different histories.
main ... 0.3

4 changed files with 78 additions and 237 deletions

1
.gitignore vendored
View File

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

View File

@ -4,7 +4,7 @@ _malias_ is a helper for mailcow instances. You can create, delete, search and l
## Installation ## Installation
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``` 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```
## Usage ## Usage
@ -14,54 +14,9 @@ For instructions run
malias -h malias -h
``` ```
## Documentation ## Screenshot
```bash ![Screenshot of malias](https://gitlab.pm/rune/malias/raw/branch/main/malias.png)
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 ## Contributing

BIN
malias.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 573 KiB

189
malias.py Normal file → Executable file
View File

@ -17,7 +17,6 @@ from string import ascii_letters, digits
from rich import print from rich import print
from argparse import RawTextHelpFormatter from argparse import RawTextHelpFormatter
from urllib.request import urlopen from urllib.request import urlopen
from operator import itemgetter
# Info pages for dev # Info pages for dev
# https://mailcow.docs.apiary.io/#reference/aliases/get-aliases/get-aliases # https://mailcow.docs.apiary.io/#reference/aliases/get-aliases/get-aliases
@ -30,8 +29,7 @@ database = filepath.joinpath('malias.db')
logfile = filepath.joinpath('malias.log') logfile = filepath.joinpath('malias.log')
Path(filepath).mkdir(parents=True, exist_ok=True) Path(filepath).mkdir(parents=True, exist_ok=True)
logging.basicConfig(filename=logfile,level=logging.INFO,format='%(message)s') logging.basicConfig(filename=logfile,level=logging.INFO,format='%(message)s')
app_version = '0.4' app_version = '0.3'
db_version = '0.2'
def get_latest_release(): def get_latest_release():
@ -43,12 +41,13 @@ def get_latest_release():
return remoteData['tag_name'] return remoteData['tag_name']
def release_check():
latest_release = get_latest_release() latest_release = get_latest_release()
def release_check():
if app_version != latest_release: if app_version != latest_release:
print('[b]New version available @ [i]https://iurl.no/malias[/b][/i]') print('[b]New version available @ [i]https://iurl.no/malias[/b][/i]')
else: else:
print ('You have the the latest version. Version: %s' %(app_version)) print ('')
def connect_database(): def connect_database():
@ -76,11 +75,6 @@ def connect_database():
alias text NOT NULL, alias text NOT NULL,
goto text NOT NULL, goto text NOT NULL,
created 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() conn.commit()
first_run(conn) first_run(conn)
@ -141,6 +135,7 @@ def set_mailserver(server):
logging.info(now + ' - Info : mailcow server updated') logging.info(now + ' - Info : mailcow server updated')
print('Your mail server has been updated.') print('Your mail server has been updated.')
conn.commit() conn.commit()
release_check()
def apikey(key): def apikey(key):
@ -150,7 +145,7 @@ def apikey(key):
logging.info(now + ' - Info : API key updated') logging.info(now + ' - Info : API key updated')
print('Your API key has been updated.') print('Your API key has been updated.')
conn.commit() conn.commit()
release_check()
def get_mail_domains(info): def get_mail_domains(info):
apikey = get_api() apikey = get_api()
@ -174,7 +169,7 @@ def get_mail_domains(info):
print('%s \t\twith %s aliases' %(remoteData[i]['domain_name'],count)) print('%s \t\twith %s aliases' %(remoteData[i]['domain_name'],count))
i+=1 i+=1
print('\n\nThere is a total of %s domains with %s aliases.' %(str(i),str(total_aliases))) print('\n\nThere is a total of %s domains with %s aliases.' %(str(i),str(total_aliases)))
release_check()
else: else:
return(remoteData) return(remoteData)
@ -202,32 +197,7 @@ def create(alias,to_address):
conn.commit() conn.commit()
logging.info(now + ' - Info : alias %s created for %s on the mailcow instance %s ' %(alias,to_address,server)) 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)) print('\n[b]Info[/b] : alias %s created for %s on the mailcow instance %s \n' %(alias,to_address,server))
release_check()
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): def delete_alias(alias):
@ -258,6 +228,7 @@ def delete_alias(alias):
else: else:
print('\n[b]Error[/b] : The alias %s not found') print('\n[b]Error[/b] : The alias %s not found')
release_check()
@ -285,6 +256,7 @@ def copy_data():
else: 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]' %(mail_server))
update_data() update_data()
release_check()
def update_data(): def update_data():
@ -331,11 +303,11 @@ def checklist(alias):
remoteData = json.loads(remote) remoteData = json.loads(remote)
i = 0 i = 0
for search in remoteData: for search in remoteData:
if alias == remoteData[i]['address'] || alias in remoteData[i]['goto']: if alias in remoteData[i]['address'] or alias in remoteData[i]['goto']:
alias_exist = True alias_exist = True
i=i+1 i=i+1
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute('SELECT count(*) FROM aliases where alias == ? or goto == ?', (alias,alias,)) cursor.execute('SELECT count(*) FROM aliases where alias like ? or goto like ?', (alias,alias,))
count = cursor.fetchone()[0] count = cursor.fetchone()[0]
if count >= 1 : if count >= 1 :
alias_exist = True alias_exist = True
@ -369,7 +341,7 @@ def list_alias():
print(the_alias + '\tgoes to\t\t' + the_goto) print(the_alias + '\tgoes to\t\t' + the_goto)
i=i+1 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))) 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)))
release_check()
def alias_id(alias): def alias_id(alias):
apikey = get_api() apikey = get_api()
@ -388,19 +360,6 @@ def alias_id(alias):
return None 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(): def number_of_aliases_in_db():
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute('SELECT count(*) FROM aliases') cursor.execute('SELECT count(*) FROM aliases')
@ -432,8 +391,6 @@ def check_local_db(alias_id):
def search(alias): def search(alias):
apikey = get_api() apikey = get_api()
mail_server = get_settings('mail_server') mail_server = get_settings('mail_server')
alias_server = number_of_aliases_on_server()
alias_db = number_of_aliases_in_db()
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute('SELECT data_copy FROM settings where id = 1') cursor.execute('SELECT data_copy FROM settings where id = 1')
result = cursor.fetchone()[0] result = cursor.fetchone()[0]
@ -448,8 +405,8 @@ def search(alias):
the_alias = remoteData[i][1].ljust(20,' ') the_alias = remoteData[i][1].ljust(20,' ')
print(the_alias + '\tgoes to\t\t' + remoteData[i][2]) print(the_alias + '\tgoes to\t\t' + remoteData[i][2])
i=i+1 i=i+1
print('\n\nData from local DB') print('\n\nData from local DB\n Run [i]malias -c to get any aliases on the server but [b]not[/b] in the local DB![/i]')
release_check()
else: else:
req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all') req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all')
req.add_header('Content-Type', 'application/json') req.add_header('Content-Type', 'application/json')
@ -465,9 +422,7 @@ def search(alias):
print(remoteData[i]['address'] + '\tgoes to\t\t' + remoteData[i]['goto']) print(remoteData[i]['address'] + '\tgoes to\t\t' + remoteData[i]['goto'])
i=i+1 i=i+1
print('\n\nData from server') print('\n\nData from server')
if alias_server != alias_db: release_check()
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(): def get_mailcow_version():
@ -479,7 +434,11 @@ def get_mailcow_version():
current = urllib.request.urlopen(req) current = urllib.request.urlopen(req)
remote = current.read().decode('utf-8') remote = current.read().decode('utf-8')
remoteData = json.loads(remote) 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) 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() tags = remote_heads.splitlines()
for x in tags: for x in tags:
string = x.rsplit('/',2) string = x.rsplit('/',2)
@ -495,7 +454,7 @@ def get_mailcow_version():
def show_current_info(): def show_current_info():
API = get_api() API = get_api()
mail_server = get_settings('mail_server') mail_server = get_settings('mail_server')
latest_release = get_latest_release()
if API == 'DUMMY_KEY': if API == 'DUMMY_KEY':
API = 'Missing API Key!' API = 'Missing API Key!'
@ -514,89 +473,26 @@ def show_current_info():
else: else:
domain = domain + str(mail_domains[i]['domain_name']) domain = domain + str(mail_domains[i]['domain_name'])
i+=1 i+=1
print('\n[b]malias[/b] - Manage aliases on Mailcow instance.') print('\n[b]malias[/b] - Manage aliases on mailcow Instance.')
print('===================================================') print('===================================================')
print("API key\t\t\t: [b]%s[/b]" % (API)) print('API key : [b]%s[/b]' % (API))
print("Mailcow Instance\t: [b]%s[/b]" % (mail_server)) print('Mailcow Instance : [b]%s[/b]' % (mail_server))
print("Active domains\t\t: [b]%s[/b]" % (domain)) print('Active domains : [b]%s[/b]' % (domain))
print("Mailcow version\t\t: [b]%s[/b]" % (mailcow_version)) print('Mailcow version : [b]%s[/b]' % (mailcow_version))
print("Logfile\t\t\t: [b]%s[/b]" % (logfile)) print('Logfile : [b]%s[/b]' % (logfile))
print("Databse\t\t\t: [b]%s[b]" % (database)) print('Databse : [b]%s[b]' % (database))
print("Aliases on server\t: [b]%s[/b]" % (aliases_server)) print('Aliases on server : [b]%s[/b]' % (aliases_server))
print("Aliases in DB\t\t: [b]%s[/b]" % (alias_db)) print('Aliases in DB : [b]%s[/b]' % (alias_db))
print("") print('')
if app_version[:3] != latest_release: if app_version != latest_release:
print( print('App version : [b]%s[/b] a new version (%s) is available @ https://iurl.no/malias' % (app_version,latest_release))
"App version\t\t\t\t: [b]%s[/b] a new version (%s) is available @ https://iurl.no/malias"
% (app_version, latest_release)
)
else: else:
print("App version\t\t: [b]%s[/b]" % (app_version)) print('App version : [b]%s[/b]' % (app_version))
print('') print('')
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() conn = connect_database()
updatedb() # Check if db needs updating and do the updating.
parser = argparse.ArgumentParser(prog='malias', 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='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!',
@ -616,17 +512,11 @@ parser.add_argument('-m', '--server', help='Add/Uppdate mailcow instance.\n\n',
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") nargs=2, metavar=('alias@domain.com', 'to@domain.com'), required=False, action="append")
parser.add_argument('-t', '--timed', help='Add new time limited alias for user on domain. One year validity\n\n',
nargs=2, metavar=('user@domain.com', 'domain.com'), required=False, action="append")
parser.add_argument('-d', '--delete', help='Delete alias.\n\n', parser.add_argument('-d', '--delete', help='Delete alias.\n\n',
nargs=1, metavar=('alias@domain.com'), required=False, action="append") nargs=1, metavar=('alias@domain.com'), required=False, action="append")
# parser.add_argument('-i', '--import', help='Show current config and appliacation info\n\n', parser.add_argument('-i', '--info', help='Show current version and config 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') required=False, action='store_true')
parser.add_argument('-c', '--copy', help='Copy alias data from mailcow server to local DB.\n\n', parser.add_argument('-c', '--copy', help='Copy alias data from mailcow server to local DB.\n\n',
@ -649,13 +539,9 @@ elif args['server']:
set_mailserver(args['server'][0][0]) set_mailserver(args['server'][0][0])
elif args['add']: elif args['add']:
create(args['add'][0][0],args['add'][0][1]) 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']: elif args['delete']:
delete_alias(args['delete'][0][0]) delete_alias(args['delete'][0][0])
elif args['import']: elif args['info']:
import_data(args['import'][0][0])
elif args['version']:
show_current_info() show_current_info()
elif args['copy']: elif args['copy']:
copy_data() copy_data()
@ -666,4 +552,5 @@ elif args['domains']:
else: 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')