rune/malias
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases 10 Activity

Compare commits

base: rune:main
Branches Tags
rune:main
rune:refactor
rune:0.4
rune:0.3.2
rune:0.3.1
rune:0.3
rune:0.2.4
rune:0.2.3
rune:0.2.2
rune:0.2.1
rune:0.2
rune:0.1
..
compare: rune:0.1
Branches Tags
rune:main
rune:refactor
rune:0.4
rune:0.3.2
rune:0.3.1
rune:0.3
rune:0.2.4
rune:0.2.3
rune:0.2.2
rune:0.2.1
rune:0.2
rune:0.1

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

4 changed files with 65 additions and 403 deletions
Show Stats Expand all files Collapse all files

4
.gitignore vendored
View File

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

51
README.md
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](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 relase from https://gitlab.pm/rune/malias/releases. Unzip and move to a folder in you path (ease of use). You can also 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,55 +14,6 @@ For instructions run
malias -h
```
## Documentation
```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
Pull requests are welcome. For major changes, please open an issue first

342
malias.py
View File

@ -10,14 +10,13 @@ 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 argparse import RawTextHelpFormatter
#from urllib import Request, urlopen
from urllib.request import urlopen
from operator import itemgetter
# Info pages for dev
# https://mailcow.docs.apiary.io/#reference/aliases/get-aliases/get-aliases
@ -30,25 +29,7 @@ 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.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))
app_version = '0.1'
def connect_database():
@ -76,11 +57,6 @@ 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)
@ -123,7 +99,6 @@ def get_settings(kind):
def get_api():
latest_release = get_latest_release()
cursor = conn.cursor()
cursor.execute('SELECT api FROM apikey')
apikey = cursor.fetchone()[0]
@ -143,6 +118,7 @@ def set_mailserver(server):
conn.commit()
def apikey(key):
now = datetime.now().strftime("%m-%d-%Y %H:%M")
cursor = conn.cursor()
@ -152,33 +128,6 @@ def apikey(key):
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')
@ -204,31 +153,6 @@ def create(alias,to_address):
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')
@ -240,6 +164,8 @@ def delete_alias(alias):
response = requests.post('https://'+server+'/api/v1/delete/alias',
data=json.dumps(data), headers=headers)
response_data = response.json()
print(response_data)
exit(0)
if response_data[0]['type'] == 'success':
if check_local_db(the_alias_id) == 1:
now = datetime.now().strftime("%m-%d-%Y %H:%M")
@ -260,7 +186,6 @@ def delete_alias(alias):
def copy_data():
apikey = get_api()
mail_server = get_settings('mail_server')
@ -283,44 +208,10 @@ def copy_data():
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\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')
print('\n[b]Info[/b] : aliases alreday imported from the mailcow instance %s to local DB\n' %(mail_server))
def checklist(alias):
alias_exist = None
apikey = get_api()
mail_server = get_settings('mail_server')
req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all')
@ -331,21 +222,14 @@ def checklist(alias):
remoteData = json.loads(remote)
i = 0
for search in remoteData:
if alias == remoteData[i]['address'] || alias in remoteData[i]['goto']:
alias_exist = True
if alias in remoteData[i]['address']:
return True
i=i+1
cursor = conn.cursor()
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
return None
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')
@ -354,21 +238,13 @@ def list_alias():
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('==================================================================')
print('\n[b]malias[/b] - All aliases on %s' %(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)
print(the_alias + '\tgoes to\t\t' + remoteData[i]['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):
@ -388,19 +264,6 @@ 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')
@ -429,73 +292,21 @@ def check_local_db(alias_id):
def search(alias):
apikey = get_api()
results = checklist(alias)
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]
if result == 1:
search_term = '%'+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 [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')
if results == True:
print('\n\nThe mail address %s exists. Using the mailcow instance : %s\n\n'%(alias,mail_server))
else:
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
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')
print('\n\nThe mail address %s [b]does not[/b] exists. Using the mailcow instance : %s\n\n'%(alias,mail_server))
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()
mail_server = get_settings('mail_server')
latest_release = get_latest_release()
if API == 'DUMMY_KEY':
API = 'Missing API Key!'
@ -504,99 +315,21 @@ def show_current_info():
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
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('\n[b]malias[/b] - Manage aliases on mailcow Instance.')
print('===================================================')
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('API key : [b]%s[/b]' % (API))
print('mailcow Instance : [b]%s[/b]' % (mail_server))
print('Logfile : [b]%s[/b]' % (logfile))
print('Aliases on server : [b]%s[/b]' % (aliases_server))
print('Aliases in DB : [b]%s[/b]' % (alias_db))
print('')
print('App version : [b]%s[/b] (https://gitlab.pm/rune/malias)' % (app_version))
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()
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!',
@ -616,17 +349,11 @@ parser.add_argument('-m', '--server', help='Add/Uppdate mailcow instance.\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('-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', '--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',
parser.add_argument('-v', '--version', help='Show current version and config info\n\n',
required=False, action='store_true')
parser.add_argument('-c', '--copy', help='Copy alias data from mailcow server to local DB.\n\n',
@ -635,9 +362,6 @@ parser.add_argument('-c', '--copy', help='Copy alias data from mailcow server to
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')
args = vars(parser.parse_args())
@ -649,20 +373,14 @@ 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['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(True)
else:

5
requirements.txt
View File

@ -11,9 +11,7 @@ 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
gpg==1.19.0
idna==3.4
importlib-metadata==6.1.0
jaraco.classes==3.2.3
@ -38,7 +36,6 @@ 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