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.3
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.3" have entirely different histories.
main ... 0.3

4 changed files with 78 additions and 237 deletions
Show Stats Expand all files Collapse all files

1
.gitignore vendored
View File

@ -1,5 +1,4 @@
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 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
@ -14,54 +14,9 @@ For instructions run
malias -h
```
## Documentation
## Screenshot
```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...
![Screenshot of malias](https://gitlab.pm/rune/malias/raw/branch/main/malias.png)
## Contributing

BIN
malias.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 573 KiB

263
malias.py Normal file → Executable file
View File

@ -17,7 +17,6 @@ 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
@ -30,8 +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'
app_version = '0.3'
def get_latest_release():
@ -43,13 +41,14 @@ def get_latest_release():
return remoteData['tag_name']
latest_release = get_latest_release()
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))
print ('')
def connect_database():
Path(filepath).mkdir(parents=True, exist_ok=True)
@ -76,18 +75,13 @@ 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()
@ -100,12 +94,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]
@ -119,9 +113,9 @@ def get_settings(kind):
return first_run_status
if kind == 'copy_status':
return copy_status
def get_api():
latest_release = get_latest_release()
cursor = conn.cursor()
@ -132,25 +126,26 @@ 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()
release_check()
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()
release_check()
def get_mail_domains(info):
apikey = get_api()
@ -174,11 +169,11 @@ def get_mail_domains(info):
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)))
release_check()
else:
return(remoteData)
def create(alias,to_address):
now = datetime.now().strftime("%m-%d-%Y %H:%M")
server = get_settings('mail_server')
@ -191,7 +186,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))
@ -202,33 +197,8 @@ 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))
release_check()
def delete_alias(alias):
server = get_settings('mail_server')
@ -238,7 +208,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:
@ -258,6 +228,7 @@ def delete_alias(alias):
else:
print('\n[b]Error[/b] : The alias %s not found')
release_check()
@ -277,7 +248,7 @@ def copy_data():
for data in remoteData:
cursor.execute('INSERT INTO aliases values(?,?,?,?)', (remoteData[i]['id'], remoteData[i]['address'],remoteData[i]['goto'],now))
i=i+1
cursor.execute('UPDATE settings SET data_copy = ? WHERE id = 1',(1,))
conn.commit()
logging.info(now + ' - Info : aliases imported from the mailcow instance %s to local DB' %(mail_server))
@ -285,6 +256,7 @@ def copy_data():
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()
release_check()
def update_data():
@ -331,15 +303,15 @@ def checklist(alias):
remoteData = json.loads(remote)
i = 0
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
i=i+1
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]
if count >= 1 :
alias_exist = True
return alias_exist
@ -369,7 +341,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)))
release_check()
def alias_id(alias):
apikey = get_api()
@ -388,19 +360,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')
@ -420,20 +379,18 @@ 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]
@ -448,8 +405,8 @@ def search(alias):
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')
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:
req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all')
req.add_header('Content-Type', 'application/json')
@ -465,11 +422,9 @@ def search(alias):
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')
release_check()
def get_mailcow_version():
apikey = get_api()
mail_server = get_settings('mail_server')
@ -479,29 +434,33 @@ 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')
latest_release = get_latest_release()
mail_server = get_settings('mail_server')
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()
@ -514,119 +473,50 @@ def show_current_info():
else:
domain = domain + str(mail_domains[i]['domain_name'])
i+=1
print('\n[b]malias[/b] - Manage aliases on Mailcow instance.')
print('\n[b]malias[/b] - Manage aliases on mailcow Instance.')
print('===================================================')
print("API key\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('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('')
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)
if app_version != latest_release:
print('App version : [b]%s[/b] a new version (%s) is available @ https://iurl.no/malias' % (app_version,latest_release))
else:
print('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,))
print('App version : [b]%s[/b]' % (app_version))
print('')
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('-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")
# 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('-i', '--info', 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',
@ -649,13 +539,9 @@ 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']:
elif args['info']:
show_current_info()
elif args['copy']:
copy_data()
@ -663,7 +549,8 @@ elif args['list']:
list_alias()
elif args['domains']:
get_mail_domains(True)
else:
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')
# 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')