mirror of
https://github.com/TakeusV1/flask_ipam.git
synced 2026-10-13 17:15:33 +02:00
DEV V1.2
This commit is contained in:
Takeus committed
1 parent
a9e8a70df5
commit
eefc23fad8
14 files changed
+406
-326
No files matched your search
+5
-263
@@ -1,24 +1,21 @@
|
||||
from datetime import datetime
|
||||
from flask import Blueprint, request, render_template, flash, redirect, url_for, session
|
||||
from flask import Blueprint, render_template, redirect, url_for
|
||||
from flask_login import login_required, current_user
|
||||
from ipaddress import IPv4Network, IPv4Address
|
||||
|
||||
from app._config import app_version
|
||||
|
||||
from app.extensions import *
|
||||
from app.models import *
|
||||
from app.forms import NewPrefixForm, NewAllocationForm, ChangeAllocationForm
|
||||
|
||||
base = Blueprint('base', __name__)
|
||||
|
||||
@base.route('/',methods=['GET','POST'])
|
||||
@base.route('/')
|
||||
def home():
|
||||
if current_user.is_authenticated:
|
||||
return redirect(url_for('base.dashboard'))
|
||||
else:
|
||||
return redirect(url_for('auth.login'))
|
||||
|
||||
@base.route('/dashboard',methods=['GET','POST'])
|
||||
@base.route('/dashboard')
|
||||
@login_required
|
||||
def dashboard():
|
||||
|
||||
@@ -42,261 +39,6 @@ def dashboard():
|
||||
db_inventory=db_inventory,
|
||||
available_alloc=available_alloc,
|
||||
db_users=db_users,
|
||||
Inventory=Inventory,
|
||||
navcolor='dark'
|
||||
)
|
||||
|
||||
@base.route('/network',methods=['GET','POST'])
|
||||
@login_required
|
||||
def networks():
|
||||
networks = Network.query.all()
|
||||
form = NewPrefixForm()
|
||||
if request.method == 'POST' and form.validate_on_submit():
|
||||
try:
|
||||
network = IPv4Network(f"{str(form.subnet.data)}/{str(form.prefix.data)}", False)
|
||||
|
||||
add_net = Network(network_address=str(network.network_address),prefix=network.prefixlen,hosts=network.num_addresses-2,description=str(form.description.data))
|
||||
db.session.add(add_net)
|
||||
db.session.commit()
|
||||
|
||||
for ip in network.hosts():
|
||||
db.session.add(Allocation(ipv4=str(ip), net_id=add_net.id))
|
||||
|
||||
db.session.commit()
|
||||
return redirect(url_for('base.networks'))
|
||||
|
||||
except:
|
||||
flash("Invalid Subnet or Prefix", 'danger')
|
||||
return redirect(url_for('base.networks'))
|
||||
elif form.errors:
|
||||
flash(form.errors, 'danger')
|
||||
|
||||
return render_template('panel/networks.html', title="Prefixes", is_networks=True, app_version=app_version, prefixes=networks, form=form, Allocation=Allocation, navcolor='dark')
|
||||
|
||||
@base.route('/network/<int:net_id>/allocations',methods=['GET','POST'])
|
||||
@login_required
|
||||
def allocations(net_id):
|
||||
|
||||
active_page = request.args.get('page', type=int)
|
||||
if active_page == None:
|
||||
active_page = 1
|
||||
|
||||
next_page = active_page+1
|
||||
prev_page = active_page-1
|
||||
|
||||
if request.args.get('items', type=int) != None:
|
||||
session['items'] = request.args.get('items', type=int)
|
||||
try:
|
||||
if session['items'] == None:
|
||||
session['items'] = 15
|
||||
except:
|
||||
session['items'] = 15
|
||||
|
||||
db_network = Network.query.filter_by(id=net_id).first()
|
||||
db_allocations = db.paginate(db.select(Allocation).filter_by(net_id=net_id), per_page=session['items'] )
|
||||
|
||||
form = NewAllocationForm()
|
||||
##if request.method == 'POST' and form.validate_on_submit():
|
||||
## network = IPv4Network(f"{str(db_network.network_address)}/{str(db_network.prefix)}")
|
||||
# Check if IPv4 Address is already allocated
|
||||
## if Allocation.query.filter_by(ipv4=str(form.ipv4.data)).first() != None:
|
||||
## flash("IPv4 Address Already Allocated", 'danger')
|
||||
## return redirect(url_for('base.allocations', net_id=net_id, page=active_page))
|
||||
## # Check if IPv4 Address is in the Network
|
||||
## if IPv4Address(form.ipv4.data) in network.hosts():
|
||||
## add_alloc = Allocation(ipv4=str(form.ipv4.data), net_id=db_network.id, hostname=str(form.host_name.data), description=str(form.host_desc.data))
|
||||
## db.session.add(add_alloc)
|
||||
## db.session.commit()
|
||||
## flash("Allocation Added", 'success')
|
||||
## return redirect(url_for('base.allocations', net_id=net_id, page=active_page))
|
||||
# If not in Network
|
||||
## flash("Invalid IPv4 Address", 'danger')
|
||||
## return redirect(url_for('base.allocations', net_id=net_id, page=active_page))
|
||||
# If Form Errors
|
||||
##elif form.errors:
|
||||
## flash(form.errors, 'danger')
|
||||
|
||||
return render_template('panel/allocations.html', title="Allocations", is_networks=True, app_version=app_version, form=form, allocations=db_allocations, network=db_network, Inventory=Inventory, active_page=active_page, next_page=next_page, prev_page=prev_page, navcolor='dark')
|
||||
|
||||
## MODAL ROUTES
|
||||
|
||||
@base.get('/network/<int:net_id>/del_modal')
|
||||
@login_required
|
||||
def network_del_modal(net_id):
|
||||
db_network = Network.query.filter_by(id=net_id).first()
|
||||
modal = {
|
||||
'title': 'Delete Network',
|
||||
'body': f'Are you sure you want to delete <b>{db_network.network_address}/{db_network.prefix}</b> network ?',
|
||||
'color': 'danger',
|
||||
'action': 'delete',
|
||||
'action_text': 'Delete',
|
||||
'action_url': url_for('base.network_delete',net_id=net_id)
|
||||
}
|
||||
return render_template('panel/_modal.html', modal=modal, net_id=net_id)
|
||||
|
||||
@base.get('/network/<int:net_id>/allocations/<string:ipv4>/del_modal')
|
||||
@login_required
|
||||
def allocation_del_modal(net_id,ipv4):
|
||||
modal = {
|
||||
'title': 'Remove Allocation',
|
||||
'body': f'Are you sure you want to remove <b>{ipv4}</b> allocation ?',
|
||||
'color': 'warning',
|
||||
'action': 'delete',
|
||||
'action_text': 'Remove',
|
||||
'action_url': url_for('base.allocation_delete',net_id=net_id,ipv4=ipv4)
|
||||
}
|
||||
return render_template('panel/_modal.html', modal=modal, net_id=net_id, ipv4=ipv4, old_page=request.args.get('page', type=int), pagination_active=True)
|
||||
|
||||
|
||||
@base.route('/network/<int:net_id>/allocations/<string:ipv4>/edit_modal', methods=['GET','POST'])
|
||||
@login_required
|
||||
def allocation_edit_modal(net_id, ipv4):
|
||||
db_alloc = Allocation.query.filter_by(ipv4=ipv4).first()
|
||||
|
||||
form = ChangeAllocationForm()
|
||||
if form.validate_on_submit() and request.method == 'POST':
|
||||
db_alloc.hostname = str(form.hostname.data)
|
||||
db_alloc.description = str(form.description.data)
|
||||
db_alloc.is_used = True
|
||||
# Special Allocation
|
||||
if int(form.is_special.data) == 1:
|
||||
# DHCP
|
||||
db_alloc.is_dhcp = True
|
||||
db_alloc.is_gateway = False
|
||||
db_alloc.hostname = None
|
||||
db_alloc.description = "DHCP Allocation"
|
||||
elif int(form.is_special.data) == 2:
|
||||
# Gateway
|
||||
db_alloc.is_dhcp = False
|
||||
db_alloc.is_gateway = True
|
||||
else:
|
||||
# None
|
||||
db_alloc.is_dhcp = False
|
||||
db_alloc.is_gateway = False
|
||||
|
||||
db.session.commit()
|
||||
flash("Allocation Updated", 'success')
|
||||
return redirect(url_for('base.allocations',net_id=net_id,page=request.args.get('page', type=int)))
|
||||
|
||||
modal = {
|
||||
'title': 'Update Allocation',
|
||||
'body': f'Change informations for <b>{db_alloc.ipv4}</b> allocation.',
|
||||
'color': 'primary',
|
||||
'modal_type': 'form',
|
||||
'action_text': 'Save',
|
||||
'form_url': url_for('base.allocation_edit_modal',net_id=db_alloc.net_id,ipv4=db_alloc.ipv4),
|
||||
}
|
||||
old_page = request.args.get('page', type=int)
|
||||
if old_page == 0:
|
||||
old_page = 1
|
||||
return render_template('panel/_modal.html', modal=modal, form=form, old_page=old_page, pagination_active=True)
|
||||
|
||||
## MODIFICATION ROUTES
|
||||
@base.get('/network/<int:net_id>/allocations/<string:ipv4>/unallocate')
|
||||
@login_required
|
||||
def allocation_delete(net_id,ipv4):
|
||||
db_alloc = Allocation.query.filter_by(net_id=net_id,ipv4=ipv4).first()
|
||||
db_alloc.is_used = False
|
||||
db_alloc.hostname = None
|
||||
db_alloc.description = None
|
||||
db_alloc.is_dhcp = False
|
||||
db_alloc.is_gateway = False
|
||||
db.session.commit()
|
||||
flash("Allocation Removed", 'warning')
|
||||
return redirect(url_for('base.allocations',net_id=net_id,page=request.args.get('page', type=int)))
|
||||
|
||||
@base.get('/network/<int:net_id>/allocations/<string:ipv4>/decom/<int:action>')
|
||||
@login_required
|
||||
def allocation_decom(net_id,ipv4,action):
|
||||
db_alloc = Allocation.query.filter_by(net_id=net_id,ipv4=ipv4).first()
|
||||
if action == 1:
|
||||
db_alloc.is_decom = True
|
||||
else:
|
||||
db_alloc.is_decom = False
|
||||
db.session.commit()
|
||||
return redirect(url_for('base.allocations',net_id=net_id,page=request.args.get('page', type=int)))
|
||||
|
||||
@base.get('/network/<int:net_id>/delete')
|
||||
@login_required
|
||||
def network_delete(net_id):
|
||||
db_network = Network.query.filter_by(id=net_id).first()
|
||||
db_alloc = Allocation.query.filter_by(net_id=db_network.id).all()
|
||||
|
||||
if db_network != None:
|
||||
if db_alloc != None:
|
||||
for alloc in db_alloc:
|
||||
db.session.delete(alloc)
|
||||
|
||||
db.session.delete(db_network)
|
||||
db.session.commit()
|
||||
flash("Network Deleted", 'warning')
|
||||
return redirect(url_for('base.networks'))
|
||||
else:
|
||||
flash("Network Not Found", 'danger')
|
||||
return redirect(url_for('base.networks'))
|
||||
|
||||
## HTMX SEARCH
|
||||
@base.post('/network/<int:net_id>/allocations/search')
|
||||
@login_required
|
||||
def allocation_search(net_id):
|
||||
search = str(request.form['search']).lower()
|
||||
|
||||
db_network = Network.query.filter_by(id=net_id).first()
|
||||
db_alloc = Allocation.query.filter_by(net_id=net_id).all()
|
||||
obj_list = []
|
||||
|
||||
# Advance Search
|
||||
if '*' in search:
|
||||
search = search.split("*")[1]
|
||||
# AND
|
||||
if ',' in search:
|
||||
for alloc in db_alloc:
|
||||
if alloc.ipv4.lower() in search.split(','):
|
||||
obj_list.append(alloc)
|
||||
# BETWEEN
|
||||
if '-' in search:
|
||||
search = search.split('-')
|
||||
if search[0].split('.')[-1] == '' or search[1].split('.')[-1] == '':
|
||||
return "Invalid(s) Number(s)"
|
||||
temp_1 = int(search[0].split('.')[-1])
|
||||
temp_2 = int(search[1].split('.')[-1])
|
||||
|
||||
for alloc in db_alloc:
|
||||
if int(alloc.ipv4.split('.')[-1]) <= temp_2 and int(alloc.ipv4.split('.')[-1]) >= temp_1:
|
||||
obj_list.append(alloc)
|
||||
# DHCP
|
||||
if 'dhcp' in search:
|
||||
for alloc in db_alloc:
|
||||
if alloc.is_dhcp:
|
||||
obj_list.append(alloc)
|
||||
|
||||
## GATEWAY
|
||||
if 'gateway' in search:
|
||||
for alloc in db_alloc:
|
||||
if alloc.is_gateway:
|
||||
obj_list.append(alloc)
|
||||
|
||||
## DECOM
|
||||
if 'decom' in search:
|
||||
for alloc in db_alloc:
|
||||
if alloc.is_decom:
|
||||
obj_list.append(alloc)
|
||||
|
||||
## DECOM
|
||||
if 'used' in search:
|
||||
for alloc in db_alloc:
|
||||
if alloc.is_used:
|
||||
obj_list.append(alloc)
|
||||
|
||||
else:
|
||||
# BASIC SEARCH
|
||||
for alloc in db_alloc:
|
||||
if alloc.ipv4.lower().startswith(search):
|
||||
obj_list.append(alloc)
|
||||
elif alloc.hostname != None:
|
||||
if alloc.hostname.lower().startswith(search):
|
||||
obj_list.append(alloc)
|
||||
elif alloc.description != None:
|
||||
if alloc.description.lower().startswith(search):
|
||||
obj_list.append(alloc)
|
||||
|
||||
return render_template('panel/_table-alloc.html',allocations=obj_list,Inventory=Inventory,network=db_network)
|
||||
)
|
||||
Reference in new issue
Block a user