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

Compare commits

base: rune:main
Branches Tags
rune:main
rune:refactor
rune:2.0
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.1
Branches Tags
rune:main
rune:refactor
rune:2.0
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.1" have entirely different histories.
main ... 0.3.1

5 changed files with 433 additions and 509 deletions
Show Stats Expand all files Collapse all files

4
.gitignore vendored
View File

@ -1,6 +1,4 @@
list_alias.py
malias.zip
malias_local.py
*.json
.DS_Store
.python-version
.DS_Store

52
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,55 +14,9 @@ For instructions run
malias -h
```
## Documentation
## Screenshot
```bash
usage: malias [-h] [-c] [-s server APIKey] [-a alias@domain.com to@domain.com] [-f alias@domain.com] [-d alias@domain.com] [-t user@domain.com domain.com] [-w alias@domain.com] [-l] [-o] [-e] [-v]
Malias is an application for adding, creating, and deleting aliases on a Mailcow instance.
Use the issues section in the git repo for any problems or suggestions. https://gitlab.pm/rune/malias
options:
-h, --help show this help message and exit
-c, --copy Copy alias data from mailcow server to local DB.
-s, --set server APIKey
Set connection information.
-a, --add alias@domain.com to@domain.com
Add new alias.
-f, --find alias@domain.com
Search for alias.
-d, --delete alias@domain.com
Delete alias.
-t, --timed user@domain.com domain.com
Add new time limited alias for user on domain.
The user@domain.com is where you want the alias to be delivered to.
The domain.com is which domain to use when creating the timed-alias.
One year validity
-w alias@domain.com, --alias alias@domain.com
List timed (temprary) aliases connected toone account.
-l, --list List all aliases on the Mailcow instance.
-o, --domains List all mail domains on the Mailcow instance.
-e, --export List all mail domains on the Mailcow instance.
-v, --version Show current version and information
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

827
malias.py Normal file → Executable file
View File

@ -1,5 +1,4 @@
#!/usr/bin/python3 -W ignore::DeprecationWarning
import sys
#!/usr/bin/python3
import sqlite3
from pathlib import Path
from sqlite3 import Error
@ -7,43 +6,39 @@ import urllib.request
import json
import logging
import argparse
import requests
import os
import time
from rich import print
from datetime import datetime, timedelta
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 operator import itemgetter
import httpx
from urllib.request import urlopen
# Info pages for dev
# https://mailcow.docs.apiary.io/#reference/aliases/get-aliases/get-aliases
# https://demo.mailcow.email/api/#/Aliases
# HTTPx ref -> https://www.python-httpx.org/
homefilepath = Path.home()
filepath = homefilepath.joinpath('.config/malias2')
database = filepath.joinpath('malias2.db')
logfile = filepath.joinpath('malias2.log')
filepath = homefilepath.joinpath('.config/malias')
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')
logging.getLogger("httpx").setLevel(logging.ERROR)
app_version = '2.0'
db_version = '2.0.0'
footer = 'malias version %s'%(app_version)
def unix_to_datetime(unix_timestamp):
return datetime.fromtimestamp(unix_timestamp)
app_version = '0.3.1'
def get_latest_release():
version = 'N/A'
req = httpx.get('https://gitlab.pm/api/v1/repos/rune/malias/releases/latest',
headers={"Content-Type": "application/json"}
)
version = req.json()['tag_name']
return version
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():
@ -52,8 +47,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)
@ -66,11 +60,13 @@ def connect_database():
finally:
if conn:
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS apikey
(id integer NOT NULL PRIMARY KEY,
api text NOT NULL)''')
c.execute('''CREATE TABLE IF NOT EXISTS settings (
id INTEGER NOT NULL PRIMARY KEY,
first_run INTEGER,
server TEXT,
apikey TEXT NOT NULL,
data_copy INTEGER
)''')
c.execute('''CREATE TABLE IF NOT EXISTS aliases
@ -78,23 +74,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)''')
c.execute('''CREATE TABLE IF NOT EXISTS dbversion
(version integer NOT NULL DEFAULT 0)''')
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()
@ -102,196 +88,270 @@ def first_run(conn):
count = cursor.fetchone()[0]
if count == 0:
logging.error(now + ' - First run!')
cursor.execute('INSERT INTO settings values(?,?,?,?,?)', (0, 1, 'dummy.server','DUMMY_KEY',0))
cursor.execute('INSERT INTO dbversion values(?)', (db_version,))
cursor.execute('INSERT INTO settings values(?,?,?,?)', (1, 1, 'dummy.server',0))
cursor.execute('INSERT INTO apikey values(?,?)', (1, 'DUMMY_KEY'))
conn.commit()
return None
def get_settings(kind):
now = datetime.now().strftime("%m-%d-%Y %H:%M")
cursor = conn.cursor()
cursor.execute('SELECT * FROM settings')
data = cursor.fetchall()
first_run_status = data[0][1] # pyright: ignore
server = data[0][2] # pyright: ignore
api_key = data[0][3] # pyright: ignore
copy_status = data[0][4] # pyright: ignore
if kind == 'connection':
if server == 'dummy.server' or api_key =='DUMMY_KEY':
print('Error: No mailcow server or API key registered. Please add with [b]malias -s [i]your.server APIKEY[/i][/b]')
data = cursor.fetchall()
first_run_status = data[0][1]
mail_server = data[0][2]
copy_status = data[0][3]
if kind == 'mail_server':
if mail_server == 'dummy.server':
print('Error: No mailcow server active. Please add one with [b]malias -m [i]your.server[/i][/b]')
exit(0)
else:
connection = {'key': api_key, 'server':server}
req = httpx.get('https://'+connection['server']+'/api/v1/get/domain/0',
headers={"Content-Type": "application/json",
'X-API-Key': connection['key']
}
)
data=req.json()
code = str(req)
if code.find('200') != -1:
return connection # pyright: ignore
else:
logging.error(now + ' - Error : Server returned error: %s, ' %(data['msg']))
print('\n [b red]Error[/b red] : Server returned error [b]%s[/b]\n\n' %(data['msg']))
exit(0)
return mail_server
if kind == 'first_run_status':
return first_run_status
if kind == 'copy_status':
return copy_status
def get_last_timed(username):
connection = get_settings('connection')
req = httpx.get('https://'+connection['server']+'/api/v1/get/time_limited_aliases/%s' %username,
headers={"Content-Type": "application/json",
'X-API-Key': connection['key']
}
)
data = req.json()
data.sort(key = itemgetter('validity'), reverse=True)
return(data[0])
def set_conection_info(server,apikey):
def get_api():
latest_release = get_latest_release()
cursor = conn.cursor()
cursor.execute('SELECT api FROM apikey')
apikey = cursor.fetchone()[0]
if apikey == 'DUMMY_KEY':
print('Missing API key. Please add with [b]malias -k [i]YOUR-API-KEY[/i][/b]')
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 = ?, apikey = ? WHERE id = 0',(server, apikey))
logging.info(now + ' - Info : Connectioninformations updated')
print('Your connection information has been updated.')
cursor.execute('UPDATE settings SET server = ? WHERE id = 1',(server,))
logging.info(now + ' - Info : mailcow server 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.')
conn.commit()
def copy_data():
connection = get_settings('connection')
if get_settings('copy_status') == 0:
now = datetime.now().strftime("%m-%d-%Y %H:%M")
cursor = conn.cursor()
req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
headers={"Content-Type": "application/json",
'X-API-Key': connection['key']
}
)
data = req.json()
if not data:
print('\n [b red]Error[/b red] : No aliases on server %s. Nothing to copy!\n\n' %(connection['server']))
exit(0)
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
for data in data:
cursor.execute('INSERT INTO aliases values(?,?,?,?)', (data['id'], data['address'],data['goto'],now))
i=i+1
cursor.execute('UPDATE settings SET data_copy = ? WHERE id = 0',(1,))
conn.commit()
logging.info(now + ' - Info : Imported %s new aliases from %s ' %(str(i),connection['server']))
print('\n[b]Info[/b] : %s aliases imported from the mailcow instance %s to local DB\n' %(i, connection['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]' %(connection['server']))
update_data()
def update_data():
connection = get_settings('connection')
if get_settings('copy_status') == 1:
now = datetime.now().strftime("%m-%d-%Y %H:%M")
req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
headers={"Content-Type": "application/json",
'X-API-Key': connection['key']
}
)
data = req.json()
i = 0
count_alias = 0
cursor = conn.cursor()
for data in data:
cursor.execute('SELECT count(*) FROM aliases where alias like ? and goto like ?', (data['address'],data['goto'],))
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]
if count >= 1 :
i+=1
else:
cursor.execute('INSERT INTO aliases values(?,?,?,?)', (data['id'], data['address'],data['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),connection['server']))
print('\n[b]Info[/b] : Local DB update with %s new aliases from %s \n' %(str(count_alias),connection['server']))
else:
print('\n[b]Info[/b] : No missing aliases from local DB \n')
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")
connection = get_settings('connection') # pyright: ignore
check = checklist(alias)
if check[0] == True:
logging.error(now + ' - Error : alias %s exists on the mailcow instance %s ' %(alias,connection['server']))
print('\n[b]Error[/b] : alias %s exists on the mailcow instance %s \n' %(alias,connection['server']))
exit(0)
elif check[1] == True:
logging.error(now + ' - Error : alias %s exists in local database.' %(alias))
print('\n[b]Error[/b] : alias %s exists in local database. \n' %(alias))
server = get_settings('mail_server')
apikey = get_api()
if checklist(alias) == True:
logging.error(now + ' - Error : alias %s exists on the mailcow instance %s ' %(alias,server))
print('\n[b]Error[/b] : alias %s exists on the mailcow instance %s \n' %(alias,server))
exit(0)
else:
try:
new_data = {'address': alias,'goto': to_address,'active': "1"}
new_data = json.dumps(new_data)
req = httpx.post('https://'+connection['server']+'/api/v1/add/alias',
headers={"Content-Type": "application/json",
'X-API-Key': connection['key']
},
data=new_data
)
except httpx.HTTPError as exc:
print(f"Error while requesting {exc.request.url!r}.")
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)
mail_id = alias_id(alias)
if mail_id == None:
logging.error(now + ' - Error : alias %s not created on the mailcow instance %s ' %(alias,connection['server']))
print('\n[b]Error[/b] : alias %s exists on the mailcow instance %s \n' %(alias,connection['server']))
logging.error(now + ' - Error : alias %s not created on the mailcow instance %s ' %(alias,server))
print('\n[b]Error[/b] : alias %s exists on the mailcow instance %s \n' %(alias,server))
else:
cursor = conn.cursor()
cursor.execute('INSERT INTO aliases values(?,?,?,?)', (mail_id, alias,to_address,now))
conn.commit()
logging.info(now + ' - Info : alias %s created for %s on the mailcow instance %s ' %(alias,to_address,connection['server']))
print('\n[b]Info[/b] : alias %s created for %s on the mailcow instance %s \n' %(alias,to_address,connection['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))
def delete_alias(alias):
server = get_settings('mail_server')
apikey = get_api()
if checklist(alias) == True:
the_alias_id = alias_id(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)
response_data = response.json()
if response_data[0]['type'] == 'success':
if check_local_db(the_alias_id) == 1:
now = datetime.now().strftime("%m-%d-%Y %H:%M")
cursor = conn.cursor()
cursor.execute('DELETE from aliases where id = ?',(the_alias_id,))
logging.info(now + ' - Info : alias %s deleted from the mailcow instance %s and Local DB' %(alias,server))
conn.commit()
print('\n[b]Info[/b] : alias %s deleted from the mailcow instance %s and local DB' %(alias,server))
else:
logging.info(now + ' - Info : alias %s deleted from the mailcow instance %s.' %(alias,server))
print('\n[b]Info[/b] : alias %s deleted from the mailcow instance %s.' %(alias,server))
else:
logging.error(now + ' - Error : alias %s NOT deleted from the mailcow instance %s. Error : %s' %(alias,server,response_data))
conn.commit()
print('\n[b]Error[/b] : alias %s NOT deleted from the mailcow instance %s. Error : %s' %(alias,server,response_data))
else:
print('\n[b]Error[/b] : The alias %s not found')
def copy_data():
apikey = get_api()
mail_server = get_settings('mail_server')
if get_settings('copy_status') == 0:
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
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\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_e = None
alias_i = None
connection = get_settings('connection')
req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
headers={"Content-Type": "application/json",
'X-API-Key': connection['key']
}
)
data = req.json()
alias_exist = None
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.loads(remote)
i = 0
for data in data:
if alias == data['address'] or alias in data['goto']:
alias_e = True
for search in remoteData:
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_i = True
alias_exist = [alias_e,alias_i]
alias_exist = True
return alias_exist
def alias_id(alias):
connection = get_settings('connection')
req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
headers={"Content-Type": "application/json",
'X-API-Key': connection['key']
}
)
data = req.json()
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')
req.add_header('X-API-Key', apikey)
current = urllib.request.urlopen(req)
remote = current.read().decode('utf-8')
remoteData = json.loads(remote)
i = 0
for data in data:
if data['address'] == alias:
return data['id']
l = 0
print('\n[b]malias[/b] - All aliases on %s ([b]*[/b] also in local db)' %(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)
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()
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.loads(remote)
i = 0
for search in remoteData:
if remoteData[i]['address'] == alias:
return remoteData[i]['id']
i=i+1
return None
@ -303,77 +363,103 @@ def number_of_aliases_in_db():
return count
def number_of_aliases_on_server():
connection = get_settings('connection')
req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
headers={"Content-Type": "application/json",
'X-API-Key': connection['key']
}
)
data = req.json()
return len(data)
def number_of_aliases_on_server():
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.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):
connection = get_settings('connection')
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()
search_term = '%'+alias+'%'
cursor.execute('SELECT alias,goto from aliases where alias like ? or goto like ?',(search_term,search_term,))
localdata = cursor.fetchall()
req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
headers={"Content-Type": "application/json",
'X-API-Key': connection['key']
}
)
remotedata = req.json()
remote = []
i=0
for data in remotedata:
if alias in remotedata[i]['address'] or alias in remotedata[i]['goto']:
remote.append((remotedata[i]['address'], remotedata[i]['goto']))
i=i+1
finallist = localdata + list(set(remote) - set(localdata))
i = 0
print('\nAliases on %s containg search term [b]%s[/b]' %(connection['server'],alias))
print('=================================================================')
for data in finallist:
the_alias = finallist[i][0].ljust(20,' ')
print(the_alias + '\tgoes to\t\t' + finallist[i][1])
i=i+1
print('\n'+footer+'\n')
def get_mail_domains(info):
connection = get_settings('connection')
cursor = conn.cursor()
req = httpx.get('https://'+connection['server']+'/api/v1/get/domain/all',
headers={"Content-Type": "application/json",
'X-API-Key': connection['key']
}
)
remoteData = req.json()
if info:
total_aliases = 0
i=0
print('\n[b]malias[/b] - All email domains on %s' %(connection['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.\n%s' %(str(i),str(total_aliases),footer))
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')
else:
return(remoteData)
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')
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():
connection = get_settings('connection')
API = get_api()
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()
mail_domains = get_mail_domains(False)
domain = ""
i=0
@ -385,236 +471,87 @@ def show_current_info():
i+=1
print('\n[b]malias[/b] - Manage aliases on mailcow Instance.')
print('===================================================')
print('API key : [b]%s[/b]' % (connection['key']))
print('Mailcow Instance : [b]%s[/b]' % (connection['server']))
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('')
if float(app_version) < float(latest_release):
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('App version : [b]%s[/b]' % (app_version))
print('')
def delete_alias(alias):
status_e = None
status_i = None
now = datetime.now().strftime("%m-%d-%Y %H:%M")
connection = get_settings('connection')
check = checklist(alias)
if check[0] == None and check[1] == None:
print('\n[b]Error[/b] : The alias %s not found')
exit(0)
if check[0] or check[1] == True:
alias_server_id = alias_id(alias)
data = {'id': alias_server_id}
delete_data = json.dumps(data)
if check[0] == True and alias_server_id != None:
req = httpx.post('https://'+connection['server']+'/api/v1/delete/alias',
headers={"Content-Type": "application/json",
'X-API-Key': connection['key']
},
data=delete_data
)
data=req.json()
code = str(req)
if code.find('200') != -1:
status_e = True
else:
status_e = None
if check[1] == True:
cursor = conn.cursor()
cursor.execute('DELETE from aliases where id = ?',(alias_server_id,))
conn.commit()
status_i = True
if status_e == True and status_i == True:
logging.info(now + ' - Info : alias %s deleted from the mailcow instance %s and Local DB' %(alias,connection['server']))
print('\n[b]Info[/b] : alias %s deleted from the mailcow instance %s and local DB' %(alias,connection['server']))
if status_e == True and status_i == None:
logging.info(now + ' - Info : alias %s deleted from the mailcow instance %s.' %(alias,connection['server']))
print('\n[b]Info[/b] : alias %s deleted from the mailcow instance %s.' %(alias,connection['server']))
if status_e == None and status_i == True:
logging.info(now + ' - Info : alias %s deleted from the local database.' %(alias))
print('\n[b]Info[/b] : alias %s deleted from the local database.' %(alias))
def create_timed(username,domain):
now = datetime.now().strftime("%m-%d-%Y %H:%M")
connection = get_settings('connection')
data = {'username': username,'domain': domain,'description': 'malias v'+app_version}
data_json = json.dumps(data)
req = httpx.post('https://'+connection['server']+'/api/v1/add/time_limited_alias',data=data_json,
headers={"Content-Type": "application/json",
'X-API-Key': connection['key']
}
)
response = json.loads(req.text)
if response[0]['type'] == 'danger' and response[0]['msg'] == 'domain_invalid':
logging.error(now + ' - Error : the domain %s does not exist.' %(domain))
print('[b][red]Error[/red][/b] : the domain %s does not exist.' %(domain))
exit(0)
if response[0]['type'] == 'danger' and response[0]['msg'] == 'access_denied':
logging.error(now + ' - Error : something went wrong. The server responded with access denied.')
print('[b][red]Error[/red][/b] : something went wrong. The server responded with [b]access denied[/b].')
exit(0)
alias = get_last_timed(username)
validity = unix_to_datetime(alias['validity'])
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,connection['server']))
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 list_alias():
now = datetime.now().strftime("%m-%d-%Y %H:%M")
connection = get_settings('connection')
cursor = conn.cursor()
req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
headers={"Content-Type": "application/json",
'X-API-Key': connection['key']
}
)
data = req.json()
i = 0
l = 0
print('\n[b]malias[/b] - All aliases on %s ([b]*[/b] also in local db)' %(connection['server']))
print('==================================================================')
for search in data:
the_alias = data[i]['address'].ljust(20,' ')
the_goto = data[i]['goto'].ljust(20,' ')
cursor.execute('SELECT count(*) FROM aliases where alias like ? or goto like ?', (data[i]['address'],data[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)
i=i+1
print('\n\nTotal number of aliases %s on instance [b]%s[/b] and %s on [b]local DB[/b].' %(str(i),connection['server'],str(l)))
print('\n'+footer)
def export_data():
connection = get_settings('connection')
cursor = conn.cursor()
req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
headers={"Content-Type": "application/json",
'X-API-Key': connection['key']
}
)
data = req.json()
with open("alias.json", "w") as outfile:
json.dump(data, outfile, ensure_ascii=False, indent=4)
def list_timed_aliases(account):
connection = get_settings('connection')
req = httpx.get('https://'+connection['server']+'/api/v1/get/time_limited_aliases/'+account,
headers={"Content-Type": "application/json",
'X-API-Key': connection['key']
}
)
data = req.json()
i = 0
print('\n[b]malias[/b] - Timed aliases on %s for %s' %(connection['server'], account))
print('==========================================================================================================')
for data in data:
the_alias = data['address'].ljust(30,' ')
the_goto = data['goto'].ljust(20,' ')
the_validity = unix_to_datetime(data['validity'])
print(the_alias + '\tgoes to\t\t' + the_goto+'Valid to: '+str(the_validity))
i=i+1
#print('\n\nTotal number of timed aliases on instance [b]%s[/b] for account.' %(connection['server'],account))
print('\n'+footer)
# For Testing purposes
def list_all():
connection = get_settings('connection')
req = httpx.get('https://'+connection['server']+'/api/v1/get/alias/all',
headers={"Content-Type": "application/json",
'X-API-Key': connection['key']
}
)
data = req.json()
print(data)
def updatedb():
# Function for updatimg DB when we have to
# 26.02.2025
# Placeholder for future updates and functions.
exit(1)
conn = connect_database()
parser = argparse.ArgumentParser(prog='malias',
description='Malias is an application for adding, creating, and deleting aliases on a Mailcow instance. \n\nUse the issues section in the git repo for any problems or suggestions. https://gitlab.pm/rune/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',
nargs=1, metavar=('APIkey'), required=False, action="append")
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',
nargs=1, metavar=('mailcow-server.tld'), required=False, action="append")
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',
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('-v', '--version', help='Show current version\n\n',
required=False, action='store_true')
parser.add_argument('-c', '--copy', help='Copy alias data from mailcow server to local DB.\n\n',
required=False, action='store_true')
parser.add_argument('-s', '--set', help='Set connection information.\n\n',
nargs=2, metavar=('server', 'APIKey'), required=False, action="append")
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('-f', '--find', help='Search for alias.\n\n',
nargs=1, metavar=('alias@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('-t', '--timed', help='Add new time limited alias for user on domain. \nThe user@domain.com is where you want the alias to be delivered to.\nThe domain.com is which domain to use when creating the timed-alias.\nOne year validity\n\n',
nargs=2, metavar=('user@domain.com', 'domain.com'), required=False, action="append")
parser.add_argument('-w', '--alias', help='List timed (temprary) aliases connected toone account.\n\n',
nargs=1, metavar=('alias@domain.com'), required=False, action="append")
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')
parser.add_argument('-e', '--export', help='List all mail domains on the Mailcow instance.\n\n',
required=False, action='store_true')
parser.add_argument('-v', '--version', help='Show current version and information\n\n',
required=False, action='store_true')
args = vars(parser.parse_args())
if args['copy']:
copy_data()
elif args['set']:
set_conection_info(args['set'][0][0],args['set'][0][1])
if args['api']:
apikey(args['api'][0][0])
elif args['search']:
search(args['search'][0][0])
elif args['server']:
set_mailserver(args['server'][0][0])
elif args['add']:
create(args['add'][0][0],args['add'][0][1])
elif args['find']:
search(args['find'][0][0])
elif args['version']:
show_current_info()
elif args['delete']:
delete_alias(args['delete'][0][0])
elif args['timed']:
create_timed(args['timed'][0][0],args['timed'][0][1])
elif args['alias']:
list_timed_aliases(args['alias'][0][0])
elif args['info']:
show_current_info()
elif args['version']:
release_check()
elif args['copy']:
copy_data()
elif args['list']:
list_alias()
elif args['domains']:
get_mail_domains(True)
elif args['export']:
export_data()
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')
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')

59
requirements.txt
View File

@ -1,13 +1,48 @@
anyio==4.8.0
certifi==2025.1.31
exceptiongroup==1.2.2
h11==0.14.0
httpcore==1.0.7
httpx==0.28.1
idna==3.10
markdown-it-py==3.0.0
aiohttp==3.8.4
aiosignal==1.3.1
async-timeout==4.0.2
attrs==22.2.0
bleach==6.0.0
build==0.10.0
certifi==2022.12.7
cffi==1.15.1
charset-normalizer==3.1.0
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
idna==3.4
importlib-metadata==6.1.0
jaraco.classes==3.2.3
keyring==23.13.1
markdown-it-py==2.2.0
mdurl==0.1.2
pygments==2.19.1
rich==13.9.4
sniffio==1.3.1
typing-extensions==4.12.2
more-itertools==9.1.0
multidict==6.0.4
notmuch==0.37
notmuch2==0.37
openai==0.27.0
packaging==23.0
pipreqs==0.4.11
pkginfo==1.9.6
promptcli==1.0.4
pycparser==2.21
Pygments==2.14.0
pyproject_hooks==1.0.0
readme-renderer==37.3
requests==2.28.2
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
webencodings==0.5.1
yarg==0.1.9
yarl==1.8.2
zipp==3.15.0