mirror of
https://github.com/vmware/vsphere-automation-sdk-python.git
synced 2024-11-29 20:09:58 -05:00
388 lines
18 KiB
Python
388 lines
18 KiB
Python
#!/usr/bin/env python
|
|
|
|
"""
|
|
* *******************************************************
|
|
* Copyright (c) VMware, Inc. 2020. All Rights Reserved.
|
|
* SPDX-License-Identifier: MIT
|
|
* *******************************************************
|
|
*
|
|
* DISCLAIMER. THIS PROGRAM IS PROVIDED TO YOU "AS IS" WITHOUT
|
|
* WARRANTIES OR CONDITIONS OF ANY KIND, WHETHER ORAL OR WRITTEN,
|
|
* EXPRESS OR IMPLIED. THE AUTHOR SPECIFICALLY DISCLAIMS ANY IMPLIED
|
|
* WARRANTIES OR CONDITIONS OF MERCHANTABILITY, SATISFACTORY QUALITY,
|
|
* NON-INFRINGEMENT AND FITNESS FOR A PARTICULAR PURPOSE.
|
|
"""
|
|
|
|
__author__ = 'VMware, Inc.'
|
|
__copyright__ = 'Copyright 2020 VMware, Inc. All rights reserved.'
|
|
__vcenter_version__ = '7.0+'
|
|
|
|
from pprint import pprint
|
|
import configparser
|
|
import os
|
|
|
|
from vmware.vapi.vsphere.client import create_vsphere_client
|
|
from samples.vsphere.common import sample_cli
|
|
from samples.vsphere.common import sample_util
|
|
from samples.vsphere.common.ssl_helper import get_unverified_session
|
|
|
|
from com.vmware.vcenter.guest_client import (CustomizationSpec,
|
|
HostnameGenerator,
|
|
LinuxConfiguration,
|
|
ConfigurationSpec,
|
|
GlobalDNSSettings,
|
|
AdapterMapping,
|
|
IPSettings,
|
|
Ipv4,
|
|
Ipv6,
|
|
Ipv6Address)
|
|
from com.vmware.vcenter.guest_client import (WindowsConfiguration,
|
|
WindowsSysprep,
|
|
Domain,
|
|
GuiUnattended,
|
|
UserData,
|
|
WindowsNetworkAdapterSettings)
|
|
|
|
|
|
class CustomizationSpecManager(object):
|
|
"""
|
|
Demonstrates create/list/get/set/delete customizationSpecs in vCenter
|
|
Sample Prerequisites: 1 vcenter, no ESXi nor VM needed
|
|
"""
|
|
|
|
def __init__(self):
|
|
parser = sample_cli.build_arg_parser()
|
|
parser.add_argument('-x', '--win_password',
|
|
action='store',
|
|
help='windows admin password to be customized')
|
|
args = sample_util.process_cli_args(parser.parse_args())
|
|
if args.win_password:
|
|
self.win_password = args.win_password
|
|
else:
|
|
self.win_password = None
|
|
session = get_unverified_session() if args.skipverification else None
|
|
self.client = create_vsphere_client(server=args.server,
|
|
username=args.username,
|
|
password=args.password,
|
|
session=session)
|
|
self.specs_svc = self.client.vcenter.guest.CustomizationSpecs
|
|
# get customization config
|
|
self.config = configparser.ConfigParser()
|
|
self.linCfgPath = os.path.join(os.path.dirname(
|
|
os.path.realpath(__file__)),
|
|
'linSpec.cfg')
|
|
self.winCfgPath = os.path.join(os.path.dirname(
|
|
os.path.realpath(__file__)),
|
|
'winSpec.cfg')
|
|
self.specsAdded = []
|
|
|
|
# common method to parse specInfo for linux/windows spec
|
|
def parseSpecInfo(self):
|
|
self.specName = self.config['CREATESPEC']['specName']
|
|
self.specDesc = self.config['CREATESPEC']['specDesc']
|
|
|
|
# common method to parse network cfg for linux/windows spec
|
|
def parseNetwork(self):
|
|
# parse macAddress
|
|
self.macAddress = self.config['NETWORK'].get('macAddress')
|
|
# parse ipv4
|
|
self.ipv4Type = self.config['NETWORK'].get('ipv4Type', 'DHCP')
|
|
if self.ipv4Type == 'STATIC':
|
|
self.ipv4_prefix = self.config['NETWORK'].getint('ipv4_prefix')
|
|
self.ipv4_gateways = self.config['NETWORK'].get('ipv4_gateways')
|
|
if self.ipv4_gateways is not None:
|
|
self.ipv4_gateways = self.ipv4_gateways.split(',')
|
|
self.ipv4_ip = self.config['NETWORK'].get('ipv4_ip')
|
|
elif (self.ipv4Type == 'DHCP' or
|
|
self.ipv4Type == 'USER_INPUT_REQUIRED'):
|
|
self.ipv4_prefix = None
|
|
self.ipv4_gateways = None
|
|
self.ipv4_ip = None
|
|
else:
|
|
raise Exception('Wrong ipv4Type "{}"'.format(self.ipv4Type))
|
|
# parse ipv6
|
|
self.ipv6Type = self.config['NETWORK'].get('ipv6Type')
|
|
if self.ipv6Type == 'STATIC':
|
|
self.ipv6_prefix = self.config['NETWORK'].getint('ipv6_prefix')
|
|
self.ipv6_gateways = self.config['NETWORK'].get('ipv6_gateways')
|
|
if self.ipv6_gateways is not None:
|
|
self.ipv6_gateways = self.ipv6_gateways.split(',')
|
|
self.ipv6_ip = self.config['NETWORK'].get('ipv6_ip')
|
|
elif ((self.ipv6Type is None) or (self.ipv4Type == 'DHCP') or
|
|
(self.ipv4Type == 'USER_INPUT_REQUIRED')):
|
|
self.ipv6_prefix = None
|
|
self.ipv6_ip = None
|
|
self.ipv6_gateways = None
|
|
else:
|
|
raise Exception('Wrong ipv6Type "{}"'.format(self.ipv6Type))
|
|
|
|
# common method to parse hostname cfg for linux/windows spec
|
|
def parseHostname(self):
|
|
# parse hostname generator type
|
|
self.hostnameType =\
|
|
self.config['HOSTNAME'].get('hostnameGeneratorType',
|
|
'VIRTUAL_MACHINE')
|
|
if (self.hostnameType == 'VIRTUAL_MACHINE' or
|
|
self.hostnameType == 'USER_INPUT_REQUIRED'):
|
|
self.prefix = None
|
|
self.fixedName = None
|
|
elif self.hostnameType == 'PREFIX':
|
|
self.prefix = self.config['HOSTNAME'].get('prefix')
|
|
self.fixedName = None
|
|
elif self.hostnameType == 'FIXED':
|
|
self.fixedName = self.config['HOSTNAME'].get('fixedName')
|
|
self.prefix = None
|
|
else:
|
|
raise Exception('Wrong hostnameGeneratorType "{}"'.format(
|
|
self.hostnameType))
|
|
|
|
# common method to parse DNS cfg for linux/windows spec
|
|
def parseDns(self):
|
|
self.globalDnsServers = self.config['DNS'].get('dnsServers')
|
|
if self.globalDnsServers is not None:
|
|
self.globalDnsServers = self.globalDnsServers.split(',')
|
|
self.globalDnsSuffixs = self.config['DNS'].get('dnsSuffixs')
|
|
if self.globalDnsSuffixs is not None:
|
|
self.globalDnsSuffixs = self.globalDnsSuffixs.split(',')
|
|
|
|
def parseWinnics(self):
|
|
self.netBiosMode = self.config['WINNICS'].get('netBiosMode')
|
|
self.dnsServers = self.config['WINNICS'].get('dnsServers')
|
|
if self.dnsServers is not None:
|
|
self.dnsServers = self.dnsServers.split(',')
|
|
self.dnsDomain = self.config['WINNICS'].get('dnsDomain')
|
|
self.winsServers = self.config['WINNICS'].get('winsServers')
|
|
if self.winsServers is not None:
|
|
self.winsServers = self.winsServers.split(',')
|
|
|
|
def parseLinuxCfg(self):
|
|
self.config.read(self.linCfgPath)
|
|
self.parseSpecInfo()
|
|
self.linSpecName = self.specName
|
|
self.parseNetwork()
|
|
self.parseHostname()
|
|
self.domainName = self.config['LINUXCONFIG'].get('domainName')
|
|
self.timezone = self.config['LINUXCONFIG'].get('timezone')
|
|
self.script_text = self.config['LINUXCONFIG'].get('script_text')
|
|
self.parseDns()
|
|
|
|
def parseWinCfg(self):
|
|
self.config.read(self.winCfgPath)
|
|
self.parseSpecInfo()
|
|
self.winSpecName = self.specName
|
|
self.parseNetwork()
|
|
self.parseHostname()
|
|
self.parseDns()
|
|
self.rebootOption = self.config['WINCONFIG'].get('rebootOption')
|
|
self.fullName = self.config['WINCONFIG'].get('fullName')
|
|
self.org = self.config['WINCONFIG'].get('organization')
|
|
self.productKey = self.config['WINCONFIG'].get('productKey')
|
|
# parse domain or workgroup
|
|
self.domainType =\
|
|
self.config['WINCONFIG'].get('domainType', 'WORKGROUP')
|
|
if self.domainType == "WORKGROUP":
|
|
self.workgroup = self.config['WINCONFIG'].get('workgroup')
|
|
self.domain = None
|
|
self.domainUser = None
|
|
self.domainPass = None
|
|
elif self.domainType == "DOMAIN":
|
|
self.workgroup = None
|
|
self.domain = self.config['WINCONFIG'].get('domain')
|
|
self.domainUser = self.config['WINCONFIG'].get('domainUser')
|
|
self.domainPass = self.config['WINCONFIG'].get('domainPass')
|
|
else:
|
|
raise Exception('Wrong domainType "{}"'.format(self.domainType))
|
|
self.autoLogon =\
|
|
self.config['WINCONFIG'].getboolean('autoLogon', False)
|
|
self.autoLogonCount =\
|
|
self.config['WINCONFIG'].getint('autoLogonCount', 0)
|
|
self.timezone = self.config['WINCONFIG'].getint('timezone', 4)
|
|
# The local Admin password
|
|
# ### WARNING: USE CLEAR TEXT IN winSpec.cfg AT YOUR OWN RISK!!! ###
|
|
# Suggested to use "--win_password" command line option instead
|
|
# It will overrid this value here
|
|
if not self.win_password:
|
|
self.win_password = self.config['WINCONFIG'].get('password')
|
|
self.gui_run_once_commands =\
|
|
self.config['WINCONFIG'].get('gui_run_once_commands')
|
|
if self.gui_run_once_commands is not None:
|
|
self.gui_run_once_commands = self.gui_run_once_commands.split(',')
|
|
self.sysprepXml = self.config['WINCONFIG'].get('sysprepXml')
|
|
self.parseWinnics()
|
|
|
|
def listCustomizationSpecs(self):
|
|
"""
|
|
List CustomizationSpecs present in vc server
|
|
"""
|
|
print("------------list--------------")
|
|
print("List Of CustomizationSpecs:")
|
|
list_of_specs = self.specs_svc.list()
|
|
self.specCount = len(list_of_specs)
|
|
pprint(list_of_specs)
|
|
|
|
def createLinuxSpec(self):
|
|
print("------------create 1 linux Customizationpec----------------")
|
|
self.parseLinuxCfg()
|
|
computerName = HostnameGenerator(prefix=self.prefix,
|
|
fixed_name=self.fixedName,
|
|
type=HostnameGenerator.Type(
|
|
self.hostnameType))
|
|
spec_linuxConfig = LinuxConfiguration(domain=self.domainName,
|
|
hostname=computerName,
|
|
time_zone=self.timezone,
|
|
script_text=self.script_text)
|
|
spec_configSpec = ConfigurationSpec(linux_config=spec_linuxConfig)
|
|
# AdapterMapping
|
|
ipv4Cfg = Ipv4(type=Ipv4.Type(self.ipv4Type), prefix=self.ipv4_prefix,
|
|
gateways=self.ipv4_gateways, ip_address=self.ipv4_ip)
|
|
if self.ipv6Type is not None:
|
|
ipv6addr = [Ipv6Address(prefix=self.ipv6_prefix,
|
|
ip_address=self.ipv6_ip)]
|
|
ipv6Cfg = Ipv6(gateways=self.ipv6_gateways, ipv6=ipv6addr,
|
|
type=Ipv6.Type(self.ipv6Type))
|
|
else:
|
|
ipv6Cfg = None
|
|
ipSettings = IPSettings(windows=None, ipv4=ipv4Cfg, ipv6=ipv6Cfg)
|
|
adapterMappingList = [AdapterMapping(adapter=ipSettings,
|
|
mac_address=self.macAddress)]
|
|
# dns_settings
|
|
dns_settings = GlobalDNSSettings(dns_servers=self.globalDnsServers,
|
|
dns_suffix_list=self.globalDnsSuffixs)
|
|
# CreateSpec
|
|
linspec_spec = CustomizationSpec(configuration_spec=spec_configSpec,
|
|
interfaces=adapterMappingList,
|
|
global_dns_settings=dns_settings)
|
|
lin_create_spec = self.specs_svc.CreateSpec(name=self.specName,
|
|
description=self.specDesc,
|
|
spec=linspec_spec)
|
|
# svc Create
|
|
self.specs_svc.create(spec=lin_create_spec)
|
|
# append it to existing list, for delete and cleanup
|
|
self.specsAdded.append(self.specName)
|
|
# list after create
|
|
self.listCustomizationSpecs()
|
|
print("----------------------------")
|
|
|
|
def createWinSpec(self):
|
|
print("------------create 1 windows CustomizationSpec----------------")
|
|
self.parseWinCfg()
|
|
# IPSettings
|
|
ipv4Cfg = Ipv4(type=Ipv4.Type(self.ipv4Type), prefix=self.ipv4_prefix,
|
|
gateways=self.ipv4_gateways, ip_address=self.ipv4_ip)
|
|
if self.ipv6Type is not None:
|
|
ipv6addr = [Ipv6Address(prefix=self.ipv6_prefix,
|
|
ip_address=self.ipv6_ip)]
|
|
ipv6Cfg = Ipv6(gateways=self.ipv6_gateways, ipv6=ipv6addr,
|
|
type=Ipv6.Type(self.ipv6Type))
|
|
else:
|
|
ipv6Cfg = None
|
|
windowsNicSettings = WindowsNetworkAdapterSettings(
|
|
net_bios_mode=self.netBiosMode, dns_servers=self.dnsServers,
|
|
dns_domain=self.dnsDomain, wins_servers=self.winsServers)
|
|
ipSettings = IPSettings(windows=windowsNicSettings,
|
|
ipv4=ipv4Cfg, ipv6=ipv6Cfg)
|
|
# AdapterMapping
|
|
adapterMappingList = [AdapterMapping(adapter=ipSettings,
|
|
mac_address=self.macAddress)]
|
|
|
|
# WindowsConfiguration
|
|
myReboot = WindowsConfiguration.RebootOption(self.rebootOption)
|
|
computerName = HostnameGenerator(prefix=self.prefix,
|
|
fixed_name=self.fixedName,
|
|
type=HostnameGenerator.Type(
|
|
self.hostnameType))
|
|
userData = UserData(computer_name=computerName,
|
|
product_key=self.productKey,
|
|
full_name=self.fullName,
|
|
organization=self.org)
|
|
myDomain = Domain(domain=self.domain, workgroup=self.workgroup,
|
|
domain_username=self.domainUser,
|
|
domain_password=self.domainPass,
|
|
type=Domain.Type(self.domainType))
|
|
guiUnattended = GuiUnattended(auto_logon_count=self.autoLogonCount,
|
|
auto_logon=self.autoLogon,
|
|
time_zone=self.timezone,
|
|
password=self.win_password)
|
|
mySysprep = WindowsSysprep(domain=myDomain,
|
|
gui_unattended=guiUnattended,
|
|
user_data=userData,
|
|
gui_run_once_commands=self.
|
|
gui_run_once_commands)
|
|
winCfg = WindowsConfiguration(reboot=myReboot,
|
|
sysprep=mySysprep,
|
|
sysprep_xml=self.sysprepXml)
|
|
# CustomizationSpec
|
|
spec_configSpec = ConfigurationSpec(windows_config=winCfg)
|
|
dns_settings = GlobalDNSSettings(dns_servers=self.globalDnsServers,
|
|
dns_suffix_list=self.globalDnsSuffixs)
|
|
winspec_spec = CustomizationSpec(configuration_spec=spec_configSpec,
|
|
interfaces=adapterMappingList,
|
|
global_dns_settings=dns_settings)
|
|
# CreateSpec
|
|
win_create_spec = self.specs_svc.CreateSpec(name=self.specName,
|
|
description=self.specDesc,
|
|
spec=winspec_spec)
|
|
# list before create
|
|
self.listCustomizationSpecs()
|
|
existingSpecCount = self.specCount
|
|
# svc Create
|
|
self.specs_svc.create(spec=win_create_spec)
|
|
# append it to existing list, for delete and cleanup
|
|
self.specsAdded.append(self.specName)
|
|
# list after create
|
|
self.listCustomizationSpecs()
|
|
print("----------------------------")
|
|
newSpecCount = self.specCount
|
|
if (newSpecCount != existingSpecCount + 1):
|
|
raise Exception('Error createSpec due to spec count={}'.
|
|
format(newSpecCount))
|
|
|
|
def getSetSpec(self):
|
|
print("-----------Get existing Spec------------")
|
|
# Get created specs, modify timezone and description of the linSpec
|
|
linSpec = self.specs_svc.get(self.linSpecName)
|
|
pprint(linSpec)
|
|
winSpec = self.specs_svc.get(self.winSpecName)
|
|
pprint(winSpec)
|
|
linSpec.spec.spec.configuration_spec.linux_config.time_zone =\
|
|
'Europe/London'
|
|
linSpec.spec.description = linSpec.spec.description +\
|
|
" modified by vapi set() method"
|
|
print("-----------Set to modify existing linSpec------------")
|
|
self.specs_svc.set(name=self.linSpecName, spec=linSpec.spec)
|
|
# Now check again if its timezone and description has changed
|
|
print("-----------Get the modified linSpec------------")
|
|
linSpec = self.specs_svc.get(self.linSpecName)
|
|
pprint(linSpec)
|
|
print("----------------------------")
|
|
|
|
def deleteSpec(self):
|
|
print("-----------Delete created spec for cleanup------------")
|
|
print("-----------before delete------------")
|
|
self.listCustomizationSpecs()
|
|
existingSpecCount = self.specCount
|
|
for specName in self.specsAdded:
|
|
self.specs_svc.delete(specName)
|
|
existingSpecCount -= 1
|
|
# list again, there should be []
|
|
print("-----------after delete------------")
|
|
self.listCustomizationSpecs()
|
|
newSpecCount = self.specCount
|
|
print("----------------------------")
|
|
if (newSpecCount != existingSpecCount):
|
|
raise Exception('Error deleteSpec due to current specCount {}!={}'.
|
|
format(newSpecCount, existingSpecCount))
|
|
|
|
|
|
def main():
|
|
myCustSpecMgr = CustomizationSpecManager()
|
|
myCustSpecMgr.listCustomizationSpecs()
|
|
myCustSpecMgr.createLinuxSpec()
|
|
myCustSpecMgr.createWinSpec()
|
|
myCustSpecMgr.getSetSpec()
|
|
myCustSpecMgr.deleteSpec()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|