forked from synackray/vcenter-netbox-sync
-
Notifications
You must be signed in to change notification settings - Fork 0
/
run.py
1601 lines (1532 loc) · 66 KB
/
run.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
"""Exports vCenter objects and imports them into Netbox via Python3"""
import asyncio
import atexit
from socket import gaierror
from datetime import date, datetime
from ipaddress import ip_network
import argparse
import aiodns
import requests
from pyVim.connect import SmartConnectNoSSL, Disconnect
from pyVmomi import vim
import settings
from logger import log
from templates.netbox import Templates
def compare_dicts(dict1, dict2, dict1_name="d1", dict2_name="d2", path=""):
"""
Compares the key value pairs of two dictionaries returns whether they match.
dict1 keys and values are compared against dict2. dict2 may have keys and
values that dict1 does not evaluate against.
:param dict1: Primary dictionary to compare against :param dict2:
:type dict1: dict
:param dict2: Dictionary being compared to by :param dict1:
:type dict2: dict
:param dict1_name: Friendly name of :param dict1: for log messages
:type dict1_name: str
:param dict2_name: Friendly name of :param dict1: for log messages
:type dict2_name: str
:param path: Used to keep state of nested dictionary traversal
:type path: str
:return: `True` if :param dict2: matches all keys and values in :param dict2: else `False`
:rtype: bool
"""
# Setup paths to track key exploration. The path parameter is used to allow
# recursive comparisions and track what's being compared.
result = True
for key in dict1.keys():
dict1_path = "{}{}[{}]".format(dict1_name, path, key)
dict2_path = "{}{}[{}]".format(dict2_name, path, key)
if key not in dict2.keys():
log.debug("%s not a valid key in %s.", dict1_path, dict2_path)
result = False
elif isinstance(dict1[key], dict) and isinstance(dict2[key], dict):
log.debug(
"%s and %s contain dictionary. Evaluating.", dict1_path,
dict2_path
)
result = compare_dicts(
dict1[key], dict2[key], dict1_name, dict2_name,
path="[{}]".format(key)
)
elif isinstance(dict1[key], list) and isinstance(dict2[key], list):
log.debug(
"%s and %s key '%s' contains list. Validating dict1 items "
"exist in dict2.", dict1_path, dict2_path, key
)
if not all([bool(item in dict2[key]) for item in dict1[key]]):
log.debug(
"Mismatch: %s value is '%s' while %s value is '%s'.",
dict1_path, dict1[key], dict2_path, dict2[key]
)
result = False
# Hack for NetBox v2.6.7 requiring integers for some values
elif key in ["status", "type"]:
if dict1[key] != dict2[key]["value"]:
log.debug(
"Mismatch: %s value is '%s' while %s value is '%s'.",
dict1_path, dict1[key], dict2_path, dict2[key]["value"]
)
result = False
elif dict1[key] != dict2[key]:
log.debug(
"Mismatch: %s value is '%s' while %s value is '%s'.",
dict1_path, dict1[key], dict2_path, dict2[key]
)
# Allow the modification of device sites by ignoring the value
if "site" in path and key == "name":
log.debug("Site mismatch is allowed. Moving on.")
else:
result = False
if result:
log.debug("%s and %s values match.", dict1_path, dict2_path)
else:
log.debug("%s and %s values do not match.", dict1_path, dict2_path)
return result
log.debug("Final dictionary compare result: %s", result)
return result
def format_ip(ip_addr):
"""
Formats IPv4 addresses and subnet to IP with CIDR standard notation.
:param ip_addr: IP address with subnet; example `192.168.0.0/255.255.255.0`
:type ip_addr: str
:return: IP address with CIDR notation; example `192.168.0.0/24`
:rtype: str
"""
ip = ip_addr.split("/")[0]
cidr = ip_network(ip_addr, strict=False).prefixlen
result = "{}/{}".format(ip, cidr)
log.debug("Converted '%s' to CIDR notation '%s'.", ip_addr, result)
return result
def format_slug(text):
"""
Format string to comply to NetBox slug acceptable pattern and max length.
:param text: Text to be formatted into an acceptable slug
:type text: str
:return: Slug of allowed characters [-a-zA-Z0-9_] with max length of 50
:rtype: str
"""
allowed_chars = (
"abcdefghijklmnopqrstuvxyzABCDEFGHIJKLMNOPQRSTUVWXYZ" # Alphabet
"01234567890" # Numbers
"_-" # Symbols
)
# Replace seperators with dash
seperators = [" ", ",", "."]
for sep in seperators:
text = text.replace(sep, "-")
# Strip unacceptable characters
text = "".join([c for c in text if c in allowed_chars])
# Enforce max length
return truncate(text, max_len=50).lower()
def format_tag(tag):
"""
Format string to comply to NetBox tag format and max length.
:param tag: The text which should be formatted
:type tag: str
:return: Tag which complies to the NetBox required tag format and max length
:rtype: str
"""
# If the tag presented is an IP address then no modifications are required
try:
ip_network(tag)
except ValueError:
# If an IP was not provided then assume fqdn
tag = tag.split(".")[0]
tag = truncate(tag, max_len=100)
return tag
def format_vcenter_conn(conn):
"""
Formats :param conn: into the expected connection string for vCenter.
This supports the use of per-host credentials without breaking previous
deployments during upgrade.
:param conn: vCenter host connection details provided in settings.py
:type conn: dict
:returns: A dictionary containing the host details and credentials
:rtype: dict
"""
try:
if bool(conn["USER"] != "" and conn["PASS"] != ""):
log.debug(
"Host specific vCenter credentials provided for '%s'.",
conn["HOST"]
)
else:
log.debug(
"Host specific vCenter credentials are not defined for '%s'.",
conn["HOST"]
)
conn["USER"], conn["PASS"] = settings.VC_USER, settings.VC_PASS
except KeyError:
log.debug(
"Host specific vCenter credential key missing for '%s'. Falling "
"back to global.", conn["HOST"]
)
conn["USER"], conn["PASS"] = settings.VC_USER, settings.VC_PASS
return conn
def is_banned_asset_tag(text):
"""
Determines whether the text is a banned asset tag through various tests.
:param text: Text to be checked against banned asset tags
:type text: str
:return: `True` if a banned asset tag else `False`
:rtype: bool
"""
# Is asset tag in banned list?
text = text.lower()
banned_tags = [
"Default string", "NA", "N/A", "None", "Null", "Unknown", " ", ""
]
banned_tags = [t.lower() for t in banned_tags]
if text in banned_tags:
result = True
# Does it exceed the max allowed length for NetBox asset tags?
elif len(text) > 50:
result = True
# Does asset tag contain all spaces?
elif text.replace(" ", "") == "":
result = True
# Apparently a "good" asset tag :)
else:
result = False
return result
def main():
"""Main function ran when the script is called directly."""
parser = argparse.ArgumentParser()
parser.add_argument(
"-c", "--cleanup", action="store_true",
help="Remove all vCenter synced objects which support tagging. This "
"is helpful if you want to start fresh or stop using this script."
)
parser.add_argument(
"-v", "--verbose", action="store_true",
help="Enable verbose output. This overrides the log level in the "
"settings file. Intended for debugging purposes only."
)
args = parser.parse_args()
if args.verbose:
log.setLevel("DEBUG")
log.debug("Log level has been overriden by the --verbose argument.")
for vc_host in settings.VC_HOSTS:
try:
start_time = datetime.now()
nb = NetBoxHandler(vc_conn=vc_host)
if args.cleanup:
nb.remove_all()
log.info(
"Completed removal of vCenter instance '%s' objects. Total "
"execution time %s.",
vc_host["HOST"], (datetime.now() - start_time)
)
else:
nb.verify_dependencies()
nb.sync_objects(vc_obj_type="datacenters")
nb.sync_objects(vc_obj_type="clusters")
nb.sync_objects(vc_obj_type="hosts")
nb.sync_objects(vc_obj_type="virtual_machines")
nb.set_primary_ips()
# Optional tasks
if settings.POPULATE_DNS_NAME:
nb.set_dns_names()
log.info(
"Completed sync with vCenter instance '%s'! Total "
"execution time %s.", vc_host["HOST"],
(datetime.now() - start_time)
)
except (ConnectionError, requests.exceptions.ConnectionError,
requests.exceptions.ReadTimeout) as err:
log.warning(
"Critical connection error occurred. Skipping sync with '%s'.",
vc_host["HOST"]
)
log.debug("Connection error details: %s", err)
continue
def queue_dns_lookups(ips):
"""
Queue handler for reverse DNS lokups.
:param ips: A list of IP addresses to queue for PTR lookup.
:type ips: list
:return: IP addresses and their respective PTR record
:rtype: dict
"""
loop = asyncio.get_event_loop()
resolver = aiodns.DNSResolver(loop=loop)
if settings.CUSTOM_DNS_SERVERS and settings.DNS_SERVERS:
resolver.nameservers = settings.DNS_SERVERS
queue = asyncio.gather(*(reverse_lookup(resolver, ip) for ip in ips))
results = loop.run_until_complete(queue)
return results
async def reverse_lookup(resolver, ip):
"""
Queries for PTR record of the IP provided with async support.
:param resolver: aiodns resolver instance set to the asyncio loop
:type resolver: aiodns.DNSResolver
:param ip: IP address to request PTR record for.
:type ip: str
:return: IP Address and its PTR record
:rtype: tuple
"""
result = (ip, "")
allowed_chars = "abcdefghijklmnopqrstuvwxyz0123456789-."
log.info("Requesting PTR record for %s.", ip)
try:
resp = await resolver.gethostbyaddr(ip)
# Make sure records comply to NetBox and DNS expected format
if all([bool(c.lower() in allowed_chars) for c in resp.name]):
result = (ip, resp.name.lower())
log.debug("PTR record for %s is '%s'.", ip, result[1])
else:
log.debug(
"Invalid characters detected in PTR record '%s'. Nulling.",
resp.name
)
except aiodns.error.DNSError as err:
log.info("Unable to find record for %s: %s", ip, err.args[1])
return result
def truncate(text="", max_len=50):
"""
Ensure a string complies to the maximum length specified.
:param text: Text to be checked for length and truncated if necessary
:type text: str
:param max_len: Max length of the returned string
:type max_len: int, optional
:return: Text in :param text: truncated to :param max_len: if necessary
:rtype: str
"""
return text if len(text) < max_len else text[:max_len]
def verify_ip(ip_addr):
"""
Verify input is expected format and checks against allowed networks.
Allowed networks can be defined in the settings IPV4_ALLOWED and
IPV6_ALLOWED variables.
:param ip_addr: IP address to check for format and whether its within allowed networks
:type ip_addr: str
:return: `True` if valid IP and within the allowed networks else `False`
:rtype: bool
"""
result = False
try:
log.debug(
"Validating IP '%s' is properly formatted and within allowed "
"networks.",
ip_addr
)
# Strict is set to false to allow host address checks
version = ip_network(ip_addr, strict=False).version
global_nets = settings.IPV4_ALLOWED if version == 4 \
else settings.IPV6_ALLOWED
# Check whether the network is within the global allowed networks
log.debug(
"Checking whether IP address '%s' is within %s.", ip_addr,
global_nets
)
net_matches = [
ip_network(ip_addr, strict=False).overlaps(ip_network(net))
for net in global_nets
]
result = any(net_matches)
except ValueError as err:
log.debug("Validation of %s failed. Received error: %s", ip_addr, err)
log.debug("IP '%s' validation returned a %s status.", ip_addr, result)
return result
class vCenterHandler:
"""
Handles vCenter connection state and object data collection
:param vc_conn: Connection details for a vCenter host defined in settings.py
:type vc_conn: dict
:param nb_api_version: NetBox API version that objects must conform to
:type nb_api_version: float
"""
def __init__(self, vc_conn, nb_api_version):
self.nb_api_version = nb_api_version
self.vc_session = None # Used to hold vCenter session state
self.vc_host = vc_conn["HOST"]
self.vc_port = vc_conn["PORT"]
self.vc_user = vc_conn["USER"]
self.vc_pass = vc_conn["PASS"]
self.tags = ["Synced", "vCenter", format_tag(self.vc_host)]
def authenticate(self):
"""Create a session to vCenter and authenticate against it"""
log.info(
"Attempting authentication to vCenter instance '%s'.",
self.vc_host
)
try:
vc_instance = SmartConnectNoSSL(
host=self.vc_host,
port=self.vc_port,
user=self.vc_user,
pwd=self.vc_pass,
)
atexit.register(Disconnect, vc_instance)
self.vc_session = vc_instance.RetrieveContent()
log.info(
"Successfully authenticated to vCenter instance '%s'.",
self.vc_host
)
except (gaierror, vim.fault.InvalidLogin, OSError) as err:
if isinstance(err, OSError):
err = "System unreachable."
err_msg = (
"Unable to connect to vCenter instance '{}' on port {}. "
"Reason: {}".format(self.vc_host, self.vc_port, err)
)
log.critical(err_msg)
raise ConnectionError(err_msg)
def create_view(self, vc_obj_type):
"""
Create a view scoped to the vCenter object type desired.
This should be called before collecting data about vCenter object types.
:param vc_obj_type: vCenter object type to extract, must be key in vc_obj_views
:type vc_obj_type: str
"""
# Mapping of object type keywords to view types
vc_obj_views = {
"datacenters": vim.Datacenter,
"clusters": vim.ClusterComputeResource,
"hosts": vim.HostSystem,
"virtual_machines": vim.VirtualMachine,
}
# Ensure an active vCenter session exists
if not self.vc_session:
log.info("No existing vCenter session found.")
self.authenticate()
return self.vc_session.viewManager.CreateContainerView(
self.vc_session.rootFolder, # View starting point
[vc_obj_views[vc_obj_type]], # Object types to look for
True # Should we recurively look into view
)
def get_objects(self, vc_obj_type):
"""
Collects vCenter objects of type and returns NetBox formated objects.
:param vc_obj_type: vCenter object type to extract, must be key in obj_type_map
:type vc_obj_type: str
:return: Extracted vCenter objects of :param vc_obj_type: in NetBox format
:rtype: dict
"""
log.info(
"Collecting vCenter %s objects.",
vc_obj_type[:-1].replace("_", " ") # Format virtual machines
)
# Mapping of vCenter object types to NetBox object types
obj_type_map = {
"datacenters": ["cluster_groups"],
"clusters": ["clusters"],
"hosts": [
"manufacturers", "device_types", "devices", "interfaces",
"ip_addresses"
],
"virtual_machines": [
"virtual_machines", "virtual_interfaces", "ip_addresses"
]
}
results = {}
# Setup use of NetBox templates
nbt = Templates(api_version=self.nb_api_version)
# Initalize keys expected to be returned
for nb_obj_type in obj_type_map[vc_obj_type]:
results.setdefault(nb_obj_type, [])
# Create vCenter view for object collection
container_view = self.create_view(vc_obj_type)
for obj in container_view.view:
try:
obj_name = obj.name
log.info(
"Collecting info about vCenter %s '%s' object.",
vc_obj_type, obj_name
)
if vc_obj_type == "datacenters":
results["cluster_groups"].append(nbt.cluster_group(
name=obj_name
))
elif vc_obj_type == "clusters":
results["clusters"].append(nbt.cluster(
name=obj_name,
ctype="VMware ESXi",
group=obj.parent.parent.name,
tags=self.tags
))
elif vc_obj_type == "hosts":
obj_manuf_name = truncate(
obj.summary.hardware.vendor, max_len=50
)
obj_model = truncate(obj.summary.hardware.model, max_len=50)
# NetBox Manufacturers and Device Types are susceptible to
# duplication as they are parents to multiple objects
# To avoid unnecessary querying we check to make sure they
# haven't already been collected
if not obj_manuf_name in [
res["name"] for res in results["manufacturers"]]:
log.debug(
"Collecting info to create NetBox manufacturers "
"object."
)
results["manufacturers"].append(nbt.manufacturer(
name=obj_manuf_name
))
else:
log.debug(
"Manufacturers object '%s' already exists. "
"Skipping.", obj_manuf_name
)
if not obj_model in [
res["model"] for res in
results["device_types"]]:
log.debug(
"Collecting info to create NetBox device_types "
"object."
)
results["device_types"].append(nbt.device_type(
manufacturer=obj_manuf_name,
model=obj_model,
part_number=obj_model,
tags=self.tags
))
else:
log.debug(
"Device Type object '%s' already exists. "
"Skipping.", obj_model
)
log.debug(
"Collecting info to create NetBox devices object."
)
# Attempt to find serial number and asset tag
hw_idents = { # Scan throw identifiers to find S/N
identifier.identifierType.key:
identifier.identifierValue
for identifier in
obj.summary.hardware.otherIdentifyingInfo
}
# Serial Number
if "EnclosureSerialNumberTag" in hw_idents.keys():
serial_number = hw_idents["EnclosureSerialNumberTag"]
elif "ServiceTag" in hw_idents.keys() \
and " " not in hw_idents["ServiceTag"]:
serial_number = hw_idents["ServiceTag"]
else:
serial_number = None
# Asset Tag
asset_tag = None
if settings.ASSET_TAGS:
try:
if "AssetTag" in hw_idents.keys():
if not is_banned_asset_tag(asset_tag):
asset_tag = hw_idents["AssetTag"].lower()
log.debug(
"Received asset tag '%s' from vCenter.",
asset_tag
)
else:
log.debug(
"Banned asset tag string. Nulling."
)
else:
log.debug(
"No asset tag detected for device '%s'.",
obj_name
)
log.debug("Final decided asset tag: %s", asset_tag)
except AttributeError as error:
log.debug(
"Received error when checking asset tag for "
" device '%s'. Error: %s",
obj_name, error
)
# Create NetBox device
results["devices"].append(nbt.device(
name=truncate(obj_name, max_len=64),
device_role="Server",
device_type=obj_model,
platform="VMware ESXi",
site="vCenter",
serial=serial_number,
asset_tag=asset_tag,
cluster=obj.parent.name,
status=int(
1 if obj.summary.runtime.connectionState ==
"connected" else 0
),
tags=self.tags
))
# Iterable object types
# Physical Interfaces
log.debug(
"Collecting info to create NetBox interfaces object."
)
for pnic in obj.config.network.pnic:
nic_name = truncate(pnic.device, max_len=64)
log.debug(
"Collecting info for physical interface '%s'.",
nic_name
)
# Try multiple methods of finding link speed
link_speed = pnic.spec.linkSpeed
if link_speed is None:
try:
link_speed = "{}Mbps ".format(
pnic.validLinkSpecification[0].speedMb
)
except IndexError:
log.debug(
"No link speed detected for physical "
"interface '%s'.", nic_name
)
else:
link_speed = "{}Mbps ".format(
pnic.spec.linkSpeed.speedMb
)
results["interfaces"].append(nbt.device_interface(
device=truncate(obj_name, max_len=64),
name=nic_name,
itype=32767, # Other
mac_address=pnic.mac,
# Interface speed is placed in the description as it
# is irrelevant to making connections and an error
# prone mapping process
description="{}Physical Interface".format(
link_speed
),
enabled=bool(link_speed),
tags=self.tags,
))
# Virtual Interfaces
for vnic in obj.config.network.vnic:
nic_name = truncate(vnic.device, max_len=64)
log.debug(
"Collecting info for virtual interface '%s'.",
nic_name
)
results["interfaces"].append(nbt.device_interface(
device=truncate(obj_name, max_len=64),
name=nic_name,
itype=0, # Virtual
mac_address=vnic.spec.mac,
mtu=vnic.spec.mtu,
tags=self.tags,
))
# IP Addresses
ip_addr = vnic.spec.ip.ipAddress
log.debug(
"Collecting info for IP Address '%s'.",
ip_addr
)
results["ip_addresses"].append(nbt.ip_address(
address="{}/{}".format(
ip_addr, vnic.spec.ip.subnetMask
),
device=truncate(obj_name, max_len=64),
interface=nic_name,
tags=self.tags,
))
elif vc_obj_type == "virtual_machines":
log.info(
"Collecting info about vCenter %s '%s' object.",
vc_obj_type, obj_name
)
# Virtual Machines
log.debug(
"Collecting info for virtual machine '%s'", obj_name
)
# Platform
vm_family = obj.guest.guestFamily
platform = None
cluster = truncate(
obj.runtime.host.parent.name, max_len=100
)
if vm_family is not None:
if "linux" in vm_family:
platform = {"name": "Linux"}
elif "windows" in vm_family:
platform = {"name": "Windows"}
results["virtual_machines"].append(nbt.virtual_machine(
name=truncate(obj_name, max_len=64),
cluster=cluster,
status=int(
1 if obj.runtime.powerState == "poweredOn" else 0
),
role="Server",
platform=platform,
memory=obj.config.hardware.memoryMB,
disk=int(sum([
comp.capacityInKB for comp in
obj.config.hardware.device
if isinstance(comp, vim.vm.device.VirtualDisk)
]) / 1024 / 1024), # Kilobytes to Gigabytes
vcpus=obj.config.hardware.numCPU,
tags=self.tags
))
# If VMware Tools is not detected then we cannot reliably
# collect interfaces and IP addresses
if vm_family:
for index, nic in enumerate(obj.guest.net):
# Interfaces
nic_name = "vNIC{}".format(index)
log.debug(
"Collecting info for virtual interface '%s'.",
nic_name
)
results["virtual_interfaces"].append(
nbt.vm_interface(
virtual_machine=obj_name,
itype=0,
name=nic_name,
mac_address=nic.macAddress,
enabled=nic.connected,
tags=self.tags
))
# IP Addresses
if nic.ipConfig is not None:
for ip in nic.ipConfig.ipAddress:
ip_addr = "{}/{}".format(
ip.ipAddress, ip.prefixLength
)
log.debug(
"Collecting info for IP Address '%s'.",
ip_addr
)
results["ip_addresses"].append(
nbt.ip_address(
address=ip_addr,
virtual_machine=obj_name,
interface=nic_name,
tags=self.tags
))
except AttributeError:
log.warning(
"Unable to collect necessary data for vCenter %s '%s'"
"object. Skipping.", vc_obj_type, obj
)
continue
container_view.Destroy()
log.debug(
"Collected %s vCenter %s object%s.", len(results),
vc_obj_type[:-1].replace("_", " "),
"s" if len(results) != 1 else "",
)
return results
class NetBoxHandler:
"""
Handles NetBox connection state and interaction with API
:param vc_conn: Connection details for a vCenter host defined in settings.py
:type vc_conn: dict
"""
def __init__(self, vc_conn):
self.nb_api_url = "http{}://{}{}/api/".format(
("s" if not settings.NB_DISABLE_TLS else ""), settings.NB_FQDN,
(":{}".format(settings.NB_PORT) if settings.NB_PORT != 443 else "")
)
self.nb_session = self._create_nb_session()
self.nb_api_version = self._get_api_version()
# NetBox object type relationships when working in the API
self.obj_map = {
"cluster_groups": {
"api_app": "virtualization",
"api_model": "cluster-groups",
"key": "name",
"prune": False,
},
"cluster_types": {
"api_app": "virtualization",
"api_model": "cluster-types",
"key": "name",
"prune": False,
},
"clusters": {
"api_app": "virtualization",
"api_model": "clusters",
"key": "name",
"prune": True,
"prune_pref": 2
},
"device_roles": {
"api_app": "dcim",
"api_model": "device-roles",
"key": "name",
"prune": False,
},
"device_types": {
"api_app": "dcim",
"api_model": "device-types",
"key": "model",
"prune": True,
"prune_pref": 3
},
"devices": {
"api_app": "dcim",
"api_model": "devices",
"key": "name",
"prune": True,
"prune_pref": 4
},
"interfaces": {
"api_app": "dcim",
"api_model": "interfaces",
"key": "name",
"prune": True,
"prune_pref": 5
},
"ip_addresses": {
"api_app": "ipam",
"api_model": "ip-addresses",
"key": "address",
"prune": True,
"prune_pref": 8
},
"manufacturers": {
"api_app": "dcim",
"api_model": "manufacturers",
"key": "name",
"prune": False,
},
"platforms": {
"api_app": "dcim",
"api_model": "platforms",
"key": "name",
"prune": False,
},
"prefixes": {
"api_app": "ipam",
"api_model": "prefixes",
"key": "prefix",
"prune": False,
},
"sites": {
"api_app": "dcim",
"api_model": "sites",
"key": "name",
"prune": True,
"prune_pref": 1
},
"tags": {
"api_app": "extras",
"api_model": "tags",
"key": "name",
"prune": False,
},
"virtual_machines": {
"api_app": "virtualization",
"api_model": "virtual-machines",
"key": "name",
"prune": True,
"prune_pref": 6
},
"virtual_interfaces": {
"api_app": "virtualization",
"api_model": "interfaces",
"key": "name",
"prune": True,
"prune_pref": 7
},
}
self.vc_tag = format_tag(vc_conn["HOST"])
self.vc = vCenterHandler(
format_vcenter_conn(vc_conn), nb_api_version=self.nb_api_version
)
def _create_nb_session(self):
"""
Creates a session with NetBox
:return: `True` if session created else `False`
:rtype: bool
"""
header = {"Authorization": "Token {}".format(settings.NB_API_KEY)}
session = requests.Session()
session.headers.update(header)
self.nb_session = session
log.info("Created new HTTP Session for NetBox.")
return session
def _get_api_version(self):
"""
Determines the current NetBox API Version
:return: NetBox API version
:rtype: float
"""
with self.nb_session.get(
self.nb_api_url, timeout=10,
verify=(not settings.NB_INSECURE_TLS)) as resp:
result = float(resp.headers["API-Version"])
log.info("Detected NetBox API v%s.", result)
return result
def get_primary_ip(self, nb_obj_type, nb_id):
"""
Collects the primary IP of a NetBox device or virtual machine.
:param nb_obj_type: NetBox object type; must match key in self.obj_map
:type nb_obj_type: str
:param nb_id: NetBox object ID of parent object where IP is configured
:type nb_id: int
:return: Primary IP and ID of the requested NetBox device or virtual machine
:rtype: dict
"""
query_key = str(
"device_id" if nb_obj_type == "devices" else "virtual_machine_id"
)
req = self.request(
req_type="get", nb_obj_type="ip_addresses",
query="?{}={}".format(query_key, nb_id)
)
log.debug("Found %s child IP addresses.", req["count"])
if req["count"] > 0:
result = {
"address": req["results"][0]["address"],
"id": req["results"][0]["id"]
}
log.debug(
"Selected %s (ID: %s) as primary IP.", result["address"],
result["id"]
)
else:
result = None
return result
def request(self, req_type, nb_obj_type, data=None, query=None, nb_id=None):
"""
HTTP requests and exception handler for NetBox
:param req_type: HTTP method type (GET, POST, PUT, PATCH, DELETE)
:type req_type: str
:param nb_obj_type: NetBox object type, must match keys in self.obj_map
:type nb_obj_type: str
:param data: NetBox object key value pairs
:type data: dict, optional
:param query: Filter for GET method requests
:type query: str, optional
:param nb_id: NetBox Object ID used when modifying an existing object
:type nb_id: int, optional
:return: Netbox objects and their corresponding data
:rtype: dict
"""
result = None
# Generate URL
url = "{}{}/{}/{}{}".format(
self.nb_api_url,
self.obj_map[nb_obj_type]["api_app"], # App that model falls under
self.obj_map[nb_obj_type]["api_model"], # Data model
query if query else "",
"{}/".format(nb_id) if nb_id else ""
)
log.debug(
"Sending %s to '%s' with data '%s'.", req_type.upper(), url, data
)
req = getattr(self.nb_session, req_type)(
url, json=data, timeout=10, verify=(not settings.NB_INSECURE_TLS)
)
# Parse status
log.debug("Received HTTP Status %s.", req.status_code)
if req.status_code == 200:
log.debug(
"NetBox %s request OK; returned %s status.", req_type.upper(),
req.status_code
)
result = req.json()
if req_type == "get":
# NetBox returns 50 results by default, this ensures all results
# are bundled together
while req.json()["next"] is not None:
url = req.json()["next"]
log.debug(
"NetBox returned more than 50 objects. Sending %s to "
"%s for additional objects.", req_type.upper(), url
)
req = getattr(self.nb_session, req_type)(url, timeout=10)
result["results"] += req.json()["results"]
elif req.status_code in [201, 204]:
log.info(
"NetBox successfully %s %s object.",
"created" if req.status_code == 201 else "deleted",
nb_obj_type,
)
elif req.status_code == 400:
if req_type == "post":
log.warning(
"NetBox failed to create %s object. A duplicate record may "
"exist or the data sent is not acceptable.", nb_obj_type
)
log.debug(
"NetBox %s status reason: %s", req.status_code, req.text
)
elif req_type == "patch":
log.warning(
"NetBox failed to modify %s object with status %s. The "
"data sent may not be acceptable.", nb_obj_type,
req.status_code