204 lines
9.0 KiB
Python
204 lines
9.0 KiB
Python
import json
|
|
from flask import Blueprint, render_template, request, redirect, url_for, flash, Response
|
|
from app.models import Client, Domain, ClientPassword, PasswordCategory
|
|
from app import db
|
|
|
|
tools_bp = Blueprint('tools', __name__)
|
|
|
|
@tools_bp.route('/tools', methods=['GET'])
|
|
def tools():
|
|
return render_template('tools.html')
|
|
|
|
|
|
@tools_bp.route('/tools/password_categories', methods=['GET', 'POST'])
|
|
def password_categories():
|
|
if request.method == 'POST':
|
|
category_name = request.form.get('category_name')
|
|
if category_name:
|
|
existing = PasswordCategory.query.filter_by(name=category_name.strip()).first()
|
|
if not existing:
|
|
new_cat = PasswordCategory(name=category_name.strip())
|
|
db.session.add(new_cat)
|
|
db.session.commit()
|
|
categories = PasswordCategory.query.all()
|
|
return render_template('password_categories.html', categories=categories)
|
|
|
|
@tools_bp.route('/tools/password_categories/delete/<int:category_id>', methods=['POST'])
|
|
def delete_password_category(category_id):
|
|
from app.models import ClientPassword, PasswordCategory
|
|
# Retrieve the category to be deleted
|
|
cat = PasswordCategory.query.get_or_404(category_id)
|
|
# Do not allow deletion of the default "General" category
|
|
if cat.name == "General":
|
|
flash("Cannot delete the default category.", "warning")
|
|
return redirect(url_for('tools.password_categories'))
|
|
# Find the default "General" category or create it if it doesn't exist
|
|
default_cat = PasswordCategory.query.filter_by(name="General").first()
|
|
if not default_cat:
|
|
default_cat = PasswordCategory(name="General")
|
|
db.session.add(default_cat)
|
|
db.session.commit()
|
|
# Reassign all passwords using the category to be deleted to "General"
|
|
ClientPassword.query.filter_by(category_id=cat.id).update({"category_id": default_cat.id})
|
|
# Delete the category
|
|
db.session.delete(cat)
|
|
db.session.commit()
|
|
flash("Category deleted and associated passwords reassigned to General.", "success")
|
|
return redirect(url_for('tools.password_categories'))
|
|
|
|
@tools_bp.route('/tools/export_clients', methods=['GET'])
|
|
def export_clients():
|
|
clients = Client.query.all()
|
|
data = []
|
|
for c in clients:
|
|
client_dict = {
|
|
'name': c.name,
|
|
'phone': c.phone,
|
|
'address': c.address,
|
|
'passwords': [],
|
|
'domains': [],
|
|
'ssl_domains': []
|
|
}
|
|
for p in c.passwords:
|
|
client_dict['passwords'].append({
|
|
'item': p.item,
|
|
'username': p.username,
|
|
'password': p.password,
|
|
'category': p.category.name if p.category else None
|
|
})
|
|
for d in c.domains:
|
|
client_dict['domains'].append({
|
|
'domain': d.domain,
|
|
'expiration_date': d.expiration_date,
|
|
'ssl_expiration_date': d.ssl_expiration_date,
|
|
'ssl_only': d.ssl_only
|
|
})
|
|
for d in c.ssl_domains:
|
|
client_dict['ssl_domains'].append({
|
|
'domain': d.domain,
|
|
'expiration_date': d.expiration_date,
|
|
'ssl_expiration_date': d.ssl_expiration_date,
|
|
'ssl_only': d.ssl_only
|
|
})
|
|
data.append(client_dict)
|
|
json_data = json.dumps(data)
|
|
response = Response(json_data, mimetype='application/json')
|
|
response.headers['Content-Disposition'] = 'attachment; filename=clients_data.json'
|
|
return response
|
|
|
|
@tools_bp.route('/tools/import_clients', methods=['GET', 'POST'])
|
|
def import_clients():
|
|
if request.method == 'POST':
|
|
file = request.files.get('file')
|
|
if file:
|
|
try:
|
|
data = json.load(file)
|
|
for client_data in data:
|
|
new_client = Client(
|
|
name = client_data.get('name'),
|
|
phone = client_data.get('phone'),
|
|
address = client_data.get('address')
|
|
)
|
|
db.session.add(new_client)
|
|
db.session.flush() # assign new_client.id
|
|
# Import passwords
|
|
for p in client_data.get('passwords', []):
|
|
category_name = p.get('category')
|
|
category = None
|
|
if category_name:
|
|
category = PasswordCategory.query.filter_by(name=category_name).first()
|
|
if not category:
|
|
category = PasswordCategory(name=category_name)
|
|
db.session.add(category)
|
|
db.session.flush()
|
|
new_password = ClientPassword(
|
|
item = p.get('item'),
|
|
username = p.get('username'),
|
|
password = p.get('password'),
|
|
client_id = new_client.id,
|
|
category = category
|
|
)
|
|
db.session.add(new_password)
|
|
# Import domain associations
|
|
for d in client_data.get('domains', []):
|
|
domain_name = d.get('domain')
|
|
if domain_name:
|
|
domain_obj = Domain.query.filter_by(domain=domain_name).first()
|
|
if not domain_obj:
|
|
domain_obj = Domain(
|
|
domain = domain_name,
|
|
expiration_date = d.get('expiration_date'),
|
|
ssl_expiration_date = d.get('ssl_expiration_date'),
|
|
ssl_only = d.get('ssl_only', False)
|
|
)
|
|
db.session.add(domain_obj)
|
|
db.session.flush()
|
|
new_client.domains.append(domain_obj)
|
|
# Import SSL domain associations
|
|
for d in client_data.get('ssl_domains', []):
|
|
domain_name = d.get('domain')
|
|
if domain_name:
|
|
domain_obj = Domain.query.filter_by(domain=domain_name).first()
|
|
if not domain_obj:
|
|
domain_obj = Domain(
|
|
domain = domain_name,
|
|
expiration_date = d.get('expiration_date'),
|
|
ssl_expiration_date = d.get('ssl_expiration_date'),
|
|
ssl_only = d.get('ssl_only', True)
|
|
)
|
|
db.session.add(domain_obj)
|
|
db.session.flush()
|
|
new_client.ssl_domains.append(domain_obj)
|
|
db.session.commit()
|
|
flash("Clients imported successfully", "success")
|
|
except Exception as e:
|
|
flash(f"Error importing clients: {str(e)}", "danger")
|
|
else:
|
|
flash("No file uploaded", "warning")
|
|
return redirect(url_for('tools.tools'))
|
|
return render_template('import_clients.html')
|
|
|
|
|
|
@tools_bp.route('/tools/export_domains', methods=['GET'])
|
|
def export_domains():
|
|
domains = Domain.query.all()
|
|
data = []
|
|
for d in domains:
|
|
data.append({
|
|
'domain': d.domain,
|
|
'expiration_date': d.expiration_date,
|
|
'ssl_expiration_date': d.ssl_expiration_date,
|
|
'ssl_only': d.ssl_only
|
|
})
|
|
json_data = json.dumps(data)
|
|
response = Response(json_data, mimetype='application/json')
|
|
response.headers['Content-Disposition'] = 'attachment; filename=domains_data.json'
|
|
return response
|
|
|
|
@tools_bp.route('/tools/import_domains', methods=['GET', 'POST'])
|
|
def import_domains():
|
|
if request.method == 'POST':
|
|
file = request.files.get('file')
|
|
if file:
|
|
try:
|
|
data = json.load(file)
|
|
for record in data:
|
|
domain_name = record.get('domain')
|
|
if domain_name:
|
|
existing = Domain.query.filter_by(domain=domain_name).first()
|
|
if not existing:
|
|
new_domain = Domain(
|
|
domain = domain_name,
|
|
expiration_date = record.get('expiration_date'),
|
|
ssl_expiration_date = record.get('ssl_expiration_date'),
|
|
ssl_only = record.get('ssl_only', False)
|
|
)
|
|
db.session.add(new_domain)
|
|
db.session.commit()
|
|
flash("Domains imported successfully", "success")
|
|
except Exception as e:
|
|
flash(f"Error importing domains: {str(e)}", "danger")
|
|
else:
|
|
flash("No file uploaded", "warning")
|
|
return redirect(url_for('tools.tools'))
|
|
return render_template('import_domains.html') |