Added timed aliases and edited README.md
This commit is contained in:
parent
1c7dd7e44d
commit
9cc931c585
1
.gitignore
vendored
1
.gitignore
vendored
@ -1,4 +1,5 @@
|
||||
list_alias.py
|
||||
malias.zip
|
||||
malias_local.py
|
||||
*.json
|
||||
.DS_Store
|
52
README.md
52
README.md
@ -14,9 +14,57 @@ 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.
|
||||
|
||||
-i alias.json, --import alias.json
|
||||
Show current config and appliacation info
|
||||
|
||||
-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
|
||||
|
||||
|
212
malias.py
Executable file → Normal file
212
malias.py
Executable file → Normal 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,8 @@ 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.3.2'
|
||||
app_version = '0.4'
|
||||
db_version = '0.2'
|
||||
|
||||
|
||||
def get_latest_release():
|
||||
@ -47,7 +49,7 @@ def release_check():
|
||||
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():
|
||||
Path(filepath).mkdir(parents=True, exist_ok=True)
|
||||
@ -74,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()
|
||||
@ -93,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]
|
||||
@ -112,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]
|
||||
@ -124,23 +132,23 @@ 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()
|
||||
|
||||
|
||||
@ -170,7 +178,7 @@ def get_mail_domains(info):
|
||||
else:
|
||||
return(remoteData)
|
||||
|
||||
|
||||
|
||||
def create(alias,to_address):
|
||||
now = datetime.now().strftime("%m-%d-%Y %H:%M")
|
||||
server = get_settings('mail_server')
|
||||
@ -183,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))
|
||||
@ -194,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')
|
||||
@ -205,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:
|
||||
@ -244,7 +277,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))
|
||||
@ -298,7 +331,7 @@ def checklist(alias):
|
||||
remoteData = json.loads(remote)
|
||||
i = 0
|
||||
for search in remoteData:
|
||||
if alias == remoteData[i]['address'] or alias == remoteData[i]['goto']:
|
||||
if alias == remoteData[i]['address'] == alias in remoteData[i]['goto']:
|
||||
alias_exist = True
|
||||
i=i+1
|
||||
cursor = conn.cursor()
|
||||
@ -306,7 +339,7 @@ def checklist(alias):
|
||||
count = cursor.fetchone()[0]
|
||||
if count >= 1 :
|
||||
alias_exist = True
|
||||
|
||||
|
||||
return alias_exist
|
||||
|
||||
|
||||
@ -355,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')
|
||||
@ -374,14 +420,14 @@ 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()
|
||||
@ -423,7 +469,7 @@ def search(alias):
|
||||
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')
|
||||
@ -437,25 +483,25 @@ def get_mailcow_version():
|
||||
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()
|
||||
@ -479,42 +525,105 @@ def show_current_info():
|
||||
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 app_version[:5] != 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('')
|
||||
|
||||
|
||||
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 config and appliacation info\n\n',
|
||||
required=False, action='store_true')
|
||||
# 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\n\n',
|
||||
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',
|
||||
@ -537,20 +646,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']:
|
||||
show_current_info()
|
||||
elif args['import']:
|
||||
import_data(args['import'][0][0])
|
||||
elif args['version']:
|
||||
release_check()
|
||||
show_current_info()
|
||||
elif args['copy']:
|
||||
copy_data()
|
||||
elif args['list']:
|
||||
list_alias()
|
||||
elif args['domains']:
|
||||
get_mail_domains(True)
|
||||
|
||||
|
||||
|
||||
|
||||
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')
|
||||
|
Loading…
Reference in New Issue
Block a user