first commit
This commit is contained in:
621
malias.py
Normal file
621
malias.py
Normal file
@@ -0,0 +1,621 @@
|
||||
#!/usr/bin/python3 -W ignore::DeprecationWarning
|
||||
import sys
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
from sqlite3 import Error
|
||||
import urllib.request
|
||||
import json
|
||||
import logging
|
||||
import argparse
|
||||
import os
|
||||
import time
|
||||
from rich import print
|
||||
from datetime import datetime, timedelta
|
||||
from string import ascii_letters, digits
|
||||
from argparse import RawTextHelpFormatter
|
||||
from operator import itemgetter
|
||||
import httpx
|
||||
|
||||
# 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')
|
||||
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)
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
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))
|
||||
|
||||
|
||||
|
||||
def connect_database():
|
||||
Path(filepath).mkdir(parents=True, exist_ok=True)
|
||||
# noinspection PyShadowingNames
|
||||
conn = None
|
||||
try:
|
||||
conn = sqlite3.connect(database)
|
||||
except Error as e:
|
||||
logging.error(time.strftime("%Y-%m-%d %H:%M") + ' - Error : ' + str(e))
|
||||
print(e)
|
||||
finally:
|
||||
if conn:
|
||||
c = conn.cursor()
|
||||
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
|
||||
(id integer NOT NULL PRIMARY KEY,
|
||||
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()
|
||||
cursor.execute('SELECT count(*) FROM settings')
|
||||
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,))
|
||||
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]')
|
||||
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)
|
||||
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):
|
||||
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.')
|
||||
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)
|
||||
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'],))
|
||||
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')
|
||||
|
||||
|
||||
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))
|
||||
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}.")
|
||||
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']))
|
||||
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']))
|
||||
|
||||
|
||||
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()
|
||||
i = 0
|
||||
for data in data:
|
||||
if alias == data['address'] or alias in data['goto']:
|
||||
alias_e = 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_i = True
|
||||
alias_exist = [alias_e,alias_i]
|
||||
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()
|
||||
i = 0
|
||||
for data in data:
|
||||
if data['address'] == alias:
|
||||
return data['id']
|
||||
i=i+1
|
||||
return None
|
||||
|
||||
|
||||
def number_of_aliases_in_db():
|
||||
cursor = conn.cursor()
|
||||
cursor.execute('SELECT count(*) FROM aliases')
|
||||
count = cursor.fetchone()[0]
|
||||
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 search(alias):
|
||||
connection = get_settings('connection')
|
||||
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))
|
||||
|
||||
else:
|
||||
return(remoteData)
|
||||
|
||||
|
||||
def show_current_info():
|
||||
connection = get_settings('connection')
|
||||
latest_release = get_latest_release()
|
||||
aliases_server = number_of_aliases_on_server()
|
||||
alias_db = number_of_aliases_in_db()
|
||||
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('===================================================')
|
||||
print('API key : [b]%s[/b]' % (connection['key']))
|
||||
print('Mailcow Instance : [b]%s[/b]' % (connection['server']))
|
||||
print('Active domains : [b]%s[/b]' % (domain))
|
||||
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):
|
||||
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',
|
||||
formatter_class=RawTextHelpFormatter,
|
||||
epilog='Making mailcow easier...')
|
||||
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])
|
||||
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['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')
|
||||
|
||||
Reference in New Issue
Block a user