Commit 161daab5 authored by Alexander Detsch's avatar Alexander Detsch 🐘
Browse files

initial commit

parents
.idea
__pycache__
*.pyc
*pw.txt
*.csv
dhcpnode.txt
This diff is collapsed.
#!/usr/bin/env python3
from dbhandler import AdminDBHandler
import argparse
import sys
import getpass
import csv
from os.path import isfile
class AdminDbCredentials:
def __init__(self, username, password,
host="admindb2-db-master.fem.tu-ilmenau.de",
port=5432,
dbname="admindb"):
self.username = username
self.password = password
self.host = host
self.port = port
self.dbname = dbname
@classmethod
def from_args(cls, args):
if not args.password:
password = getpass.getpass(prompt='AdminDB2-Password: ')
else:
password = args.password
return cls(args.user, password, args.host, args.port, args.dbname)
def search_nested(nlist, ap_name):
for ap in nlist:
if ap['ap-name'] == ap_name:
return ap['switchport-id']
return -1
def prepare_switchport_in_admindb(
switchportid,
ap_name,
db_username,
db_password,
db_host="admindb2-db-master.fem.tu-ilmenau.de",
db_port=5432,
db_dbname="admindb",
wiredswitchid=298):
db = AdminDBHandler()
if not db.connect(db_host, db_port, db_dbname, db_username, db_password, False):
raise Exception('Failed to Connect to AdminDB2!')
switchport = db.admindb_get_switchport_by_id(switchportid)
if not switchport:
db.rollback()
db.disconnect()
raise Exception('Failed to get switchport!')
userport = db.admindb_get_userport_by_switchportid(switchportid)
if not userport:
db.rollback()
db.disconnect()
raise Exception('Failed to userport associated with switchport!')
switchport = switchport[0]
userport = userport[0]
newswitchportid = db.admindb_switchport_insert_or_update(
None,
wiredswitchid,
ap_name,
ap_name,
True,
None,
2,
True,
switchport['t_vlan_id_untagged'])
if not newswitchportid:
db.rollback()
db.disconnect()
raise Exception('Failed to create virtual switchport for wired devices!')
newswitchportid = newswitchportid[0][0]
updateresult = db.admindb_userport_insert_or_update(
userport['t_id'],
userport['t_block_id'],
userport['t_room'],
'Aruba-AP',
'',
'')
if not updateresult:
db.rollback()
db.disconnect()
raise Exception('Failed to update old userport!')
patchresult = db.admindb_patch_userport_delete(switchport['t_id'], userport['t_id'])
if not patchresult:
db.rollback()
db.disconnect()
raise Exception('Failed to unpatch old userport from old switchport!')
patchresult = db.admindb_patch_userport_insert(newswitchportid, userport['t_id'])
if not patchresult:
db.rollback()
db.disconnect()
raise Exception('Failed to patch old userport to new switchport!')
newuserportid = db.admindb_userport_insert_or_update(
None,
userport['t_block_id'],
'WLAN-AP ' + userport['t_room'],
userport['t_socket'],
userport['t_panelport'],
userport['t_notice'])
if not newuserportid:
db.rollback()
db.disconnect()
raise Exception('Failed to create new userport!')
newuserportid = newuserportid[0][0]
patchresult = db.admindb_patch_userport_insert(switchport['t_id'], newuserportid)
if not patchresult:
db.rollback()
db.disconnect()
raise Exception('Failed to patch new userport to old switchport!')
updateresult = db.admindb_switchport_insert_or_update(
switchport['t_id'],
switchport['t_switch_id'],
switchport['t_name'],
switchport['t_port'],
switchport['t_useable'],
switchport['t_notice'],
12,
True,
switchport['t_vlan_id_untagged'])
if not updateresult:
db.rollback()
db.disconnect()
raise Exception('Failed to update old switchport to autoconf!')
db.commit()
db.disconnect()
return {
"switchportid": switchportid,
"userportid": userport['t_id'],
"switchportid_new": newswitchportid,
"userportid_new": newuserportid
}
def create_parser():
parser = argparse.ArgumentParser(description='Python-Script prepare a Switchport for an Aruba-APs in AdminDB2 '
'and to create a virtual Switchport for wired devices '
'behind the Aruba-AP.')
parser.add_argument('--host', type=str, default='admindb2-db-master.fem.tu-ilmenau.de',
help='[Optional] AdminDB2-DB-Host')
parser.add_argument('--port', type=int, default=5432,
help='[Optional] AdminDB2-DB-Port')
parser.add_argument('--dbname', type=str, default='admindb',
help='[Optional] AdminDB2-Database-Name')
parser.add_argument('--user', '-u', type=str,
help='AdminDB2-User')
parser.add_argument('--password', '-p', type=str,
help='AdminDB2-Password (if not given you will be asked)')
parser.add_argument('--apname', '-n', type=str,
help='Name of the Aruba-AP')
parser.add_argument('--switchportid', '-s', type=int,
help='ID of the Switchport, ignored in lookup-mode')
parser.add_argument('--skipprefix', action='store_true',
help="[Optional] Skips adding prefix 'aruba-ap-' to ap-name if not already present")
parser.add_argument('--wiredswitchid', type=int, default=298,
help='[Optional] Switch-ID of the arubawlanwired Switch')
parser.add_argument('--lookup', '-l', type=str,
help='[Lookup-Mode][Optional] path to csv; expected column names: ap-name, switchport-id / semicolon / with header')
return parser
def assert_required_args(args):
if not args.user:
raise Exception('No AdminDB2-User given!')
if not args.apname:
raise Exception('No AP-Name given!')
if args.lookup:
if not isfile(args.lookup):
raise Exception(f"file {args.lookup} not found.")
def read_ap_data_from_csv(file_name):
lookup_list = []
with open(file_name, "r") as csvfile:
csv_reader = csv.DictReader(csvfile, delimiter=";")
for row in csv_reader:
lookup_list.append({
'ap-name': row['ap-name'],
'switchport-id': row['switchport-id']
})
return lookup_list
def main():
parser = create_parser()
args = parser.parse_args()
try:
assert_required_args(args)
except Exception as e:
print("Invalid arguments: " + str(e))
parser.print_help()
return 1
db_credentials = None
try:
db_credentials = AdminDbCredentials.from_args(args)
except:
print("Missing Credentials!")
return 1
if args.lookup:
lookup_list = read_ap_data_from_csv(args.lookup)
search_result = search_nested(lookup_list, args.apname)
if search_result == -1:
print("Switchport-ID not found in lookup table.")
switchportid = int(input("Please enter the Switchport-ID: "))
else:
switchportid = search_result
else:
if not args.switchportid:
print('No Switchport-ID given!')
switchportid = int(input("Please enter the Switchport-ID: "))
switchportid = args.switchportid
# add prefix if needed (after lookup!)
if not args.skipprefix and not args.apname.startswith('aruba-ap-'):
ap_name = f"aruba-ap-{args.apname}"
try:
admindb_results = prepare_switchport_in_admindb(
switchportid,
ap_name,
db_username=db_credentials.username,
db_password=db_credentials.password,
db_host=db_credentials.host,
db_port=db_credentials.port,
db_dbname=db_credentials.dbname,
wiredswitchid=args.wiredswitchid)
print("Switchport successfully prepared!")
print(f"Switchport-ID: {admindb_results['switchportid']}")
print(f"Userport-ID: {admindb_results['userportid']}")
print(f"Switchport-ID of new virtual Switchport: {admindb_results['switchportid_new']}")
print(f"Userport-ID of userport for the Aruba-AP: {admindb_results['userportid_new']}")
except Exception as e:
print(f"Failed to prepare switchport #{switchportid} for AP {ap_name}:\n{e}")
return 1
return 0
if __name__ == '__main__':
sys.exit(main())
netaddr
requests
dokuwiki
git+https://gitlab.fem-net.de/admindb/dbhandler.git
# Windows only
#pysocks
import setuptools
setuptools.setup(
name="arubawlan_python_scripts",
version="0.0.3",
author="FeM",
author_email="technik@fem.tu-ilmenau.de",
description="Aruba-WLAN Python Scripts",
url="https://gitlab.fem-net.de/aruba-wlan/arubawlan-python-scripts",
project_urls={
"Git": "https://gitlab.fem-net.de/aruba-wlan/arubawlan-python-scripts",
},
classifiers=[
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
],
packages=setuptools.find_packages(),
python_requires=">=3.6",
install_requires=[
"netaddr",
"requests",
"dokuwiki",
"dbhandler"
],
dependency_links = [
"https://gitlab.fem-net.de/admindb/dbhandler/-/archive/master/dbhandler-master.tar.gz"
]
)
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment