308 lines
10 KiB
Python
308 lines
10 KiB
Python
"""
|
|
Generates a library ready to be used as a VCSP endpoint for content library 2016 (vsphere 6.5) and beyond.
|
|
Adapted from https://github.com/lamw/vmware-scripts/blob/master/python/make_vcsp_2022.py
|
|
"""
|
|
|
|
__author__ = 'VMware, Inc.'
|
|
__copyright__ = 'Copyright 2019 VMware, Inc. All rights reserved.'
|
|
|
|
import argparse
|
|
import datetime
|
|
import hashlib
|
|
import logging
|
|
import json
|
|
import os
|
|
import uuid
|
|
import sys
|
|
|
|
import urllib.parse
|
|
|
|
VCSP_VERSION = 2
|
|
ISO_FORMAT = "%Y-%m-%dT%H:%MZ"
|
|
FORMAT = "json"
|
|
FILE_EXTENSION_CERT = ".cert"
|
|
LIB_FILE = ''.join(("lib", os.extsep, FORMAT))
|
|
ITEMS_FILE = ''.join(("items", os.extsep, FORMAT))
|
|
ITEM_FILE = ''.join(("item", os.extsep, FORMAT))
|
|
VCSP_TYPE_OVF = "vcsp.ovf"
|
|
VCSP_TYPE_ISO = "vcsp.iso"
|
|
VCSP_TYPE_OTHER = "vcsp.other"
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def _md5_for_file(f, md5=None, block_size=2**20):
|
|
if md5 is None:
|
|
md5 = hashlib.md5()
|
|
while True:
|
|
data = f.read(block_size)
|
|
if not data:
|
|
break
|
|
md5.update(data)
|
|
return md5
|
|
|
|
|
|
def _md5_for_folder(folder):
|
|
md5 = None
|
|
for files in os.listdir(folder):
|
|
if ITEM_FILE not in files:
|
|
with open(os.path.join(folder, files), "rb") as handle:
|
|
md5 = _md5_for_file(handle, md5)
|
|
return md5.hexdigest()
|
|
|
|
|
|
def _make_lib(name, id=uuid.uuid4(), creation=datetime.datetime.now(), version=1):
|
|
return {
|
|
"vcspVersion": str(VCSP_VERSION),
|
|
"version": str(version),
|
|
"contentVersion": "1",
|
|
"name": name,
|
|
"id": "urn:uuid:%s" % id,
|
|
"created": creation.strftime(ISO_FORMAT),
|
|
"capabilities": {
|
|
"transferIn": [ "httpGet" ],
|
|
"transferOut": [ "httpGet" ],
|
|
},
|
|
"itemsHref": ITEMS_FILE
|
|
}
|
|
|
|
|
|
def _make_item(directory, vcsp_type, name, files, description="", properties={},
|
|
identifier=uuid.uuid4(), creation=datetime.datetime.now(), version=2,
|
|
library_id="", is_vapp_template="false"):
|
|
'''
|
|
add type adapter metadata for OVF template
|
|
'''
|
|
if len(name) > 80: #max size of name is 80 chars
|
|
extension = name.rsplit(".")[-1]
|
|
name = name[0:80-len(extension)-1]+"."+extension
|
|
if "urn:uuid:" not in str(identifier):
|
|
item_id = "urn:uuid:%s" % identifier
|
|
else:
|
|
item_id = identifier
|
|
type_metadata = None
|
|
if vcsp_type == VCSP_TYPE_OVF:
|
|
# generate sample type metadata for OVF template so that subscriber can show OVF VM type
|
|
type_metadata_value = "{'id':'%s','version':'%s','libraryIdParent':'%s','isVappTemplate':'%s','vmTemplate':null,'vappTemplate':null,'networks':[],'storagePolicyGroups':null}" % (item_id, str(version), library_id, is_vapp_template)
|
|
type_metadata = {
|
|
"key": "type-metadata",
|
|
"value": type_metadata_value,
|
|
"type": "String",
|
|
"domain": "SYSTEM",
|
|
"visibility": "READONLY"
|
|
}
|
|
if type_metadata:
|
|
return {
|
|
"created": creation.strftime(ISO_FORMAT),
|
|
"description": description,
|
|
"version": str(version),
|
|
"files": files,
|
|
"id": item_id,
|
|
"name": name,
|
|
"metadata": [type_metadata],
|
|
"properties": properties,
|
|
"selfHref": "%s/%s" % (urllib.parse.quote(directory), urllib.parse.quote(ITEM_FILE)),
|
|
"type": vcsp_type
|
|
}
|
|
else:
|
|
return {
|
|
"created": creation.strftime(ISO_FORMAT),
|
|
"description": description,
|
|
"version": str(version),
|
|
"files": files,
|
|
"id": item_id,
|
|
"name": name,
|
|
"properties": properties,
|
|
"selfHref": "%s/%s" % (urllib.parse.quote(directory), urllib.parse.quote(ITEM_FILE)),
|
|
"type": vcsp_type
|
|
}
|
|
|
|
|
|
def _make_items(items, version=1):
|
|
return {
|
|
"items": items
|
|
}
|
|
|
|
|
|
def _dir2item(path, directory, md5_enabled, lib_id):
|
|
files_items = []
|
|
name = os.path.split(path)[-1]
|
|
vcsp_type = VCSP_TYPE_OTHER
|
|
folder = ""
|
|
folder_md5 = ""
|
|
is_vapp = ""
|
|
for f in os.listdir(path):
|
|
if f == ".DS_Store" or f == ''.join((directory, os.extsep, FORMAT)):
|
|
continue
|
|
else:
|
|
if f == "item.json":
|
|
continue # skip the item.json meta data files
|
|
p = os.path.join(path, f)
|
|
m = hashlib.md5()
|
|
new_folder = os.path.dirname(p)
|
|
if new_folder != folder: # new folder (ex: template1/)
|
|
if md5_enabled:
|
|
folder_md5 = _md5_for_folder(new_folder)
|
|
folder = new_folder
|
|
if md5_enabled:
|
|
m.update(os.path.dirname(p).encode('utf-8'))
|
|
if ".ovf" in p:
|
|
vcsp_type = VCSP_TYPE_OVF
|
|
# TODO: ready ovf descriptor for type metadata
|
|
is_vapp = "false"
|
|
elif ".iso" in p:
|
|
vcsp_type = VCSP_TYPE_ISO
|
|
size = os.path.getsize(p)
|
|
href = "%s/%s" % (directory, f)
|
|
h = ""
|
|
if md5_enabled:
|
|
with open(p, "rb") as handle:
|
|
h = _md5_for_file(handle)
|
|
files_items.append({
|
|
"name": f,
|
|
"size": size,
|
|
"etag": folder_md5,
|
|
"hrefs": [ urllib.parse.quote(href,safe="/")]
|
|
})
|
|
return _make_item(name, vcsp_type, name, files_items, identifier = uuid.uuid4(), library_id=lib_id, is_vapp_template=is_vapp)
|
|
|
|
|
|
def make_vcsp(lib_name, lib_path, md5_enabled):
|
|
lib_json_loc = os.path.join(lib_path, LIB_FILE)
|
|
lib_items_json_loc = os.path.join(lib_path, ITEMS_FILE)
|
|
|
|
lib_id = uuid.uuid4()
|
|
lib_create = datetime.datetime.now()
|
|
lib_version = 1
|
|
updating_lib = False
|
|
if os.path.isfile(lib_json_loc):
|
|
logger.info("%s already exists (%s)" % (LIB_FILE, lib_json_loc))
|
|
try:
|
|
with open(lib_json_loc, "r") as f:
|
|
old_lib = json.load(f)
|
|
if "id" in old_lib:
|
|
lib_id = old_lib["id"].split(":")[-1]
|
|
if "created" in old_lib:
|
|
lib_create = datetime.datetime.strptime(old_lib["created"], ISO_FORMAT)
|
|
if "version" in old_lib:
|
|
lib_version = old_lib["version"]
|
|
updating_lib = True
|
|
except:
|
|
logger.error("Failed to read %s" % lib_json_loc)
|
|
pass
|
|
|
|
old_items = {}
|
|
if os.path.isfile(lib_items_json_loc):
|
|
logger.info("%s already exists (%s)" % (ITEMS_FILE, lib_items_json_loc))
|
|
try:
|
|
with open(lib_items_json_loc, "r") as f:
|
|
old_data = json.load(f)
|
|
for item in old_data["items"]:
|
|
old_items[item["name"]] = item
|
|
except:
|
|
logger.error("Failed to read %s" % lib_items_json_loc)
|
|
pass
|
|
|
|
items = []
|
|
changed = False
|
|
for item_path in os.listdir(lib_path):
|
|
p = os.path.join(lib_path, item_path)
|
|
if not os.path.isdir(p):
|
|
continue # not interesting
|
|
item_json = _dir2item(p, item_path, md5_enabled, "urn:uuid:%s" % lib_id)
|
|
if item_path not in old_items and updating_lib:
|
|
changed = True
|
|
elif item_path in old_items:
|
|
file_changed = False
|
|
item_json["id"] = old_items[item_path]["id"]
|
|
item_json["created"] = old_items[item_path]["created"]
|
|
item_json["version"] = old_items[item_path]["version"]
|
|
file_names = set([i["name"] for i in item_json["files"]])
|
|
old_file_names = set([i["name"] for i in old_items[item_path]["files"]])
|
|
if file_names != old_file_names:
|
|
# files added or removed
|
|
changed = True
|
|
file_changed = True
|
|
for f in item_json["files"]:
|
|
if file_changed:
|
|
break
|
|
for old_f in old_items[item_path]["files"]:
|
|
if f["name"] == old_f["name"] and f["etag"] != old_f["etag"]:
|
|
changed = True
|
|
file_changed = True
|
|
break
|
|
if file_changed:
|
|
item_version = int(item_json["version"])
|
|
item_json["version"] = str(item_version + 1)
|
|
del old_items[item_path]
|
|
json_item_file = ''.join((p, os.sep, ITEM_FILE))
|
|
with open(json_item_file, "w") as f:
|
|
json.dump(item_json, f, indent=2)
|
|
items.append(item_json)
|
|
|
|
if updating_lib and len(old_items) != 0:
|
|
changed = True # items were removed
|
|
|
|
if updating_lib and not changed:
|
|
logger.info("Nothing to update, quitting")
|
|
return
|
|
if changed:
|
|
lib_version = int(lib_version)
|
|
lib_version += 1
|
|
logger.info("Saving results to %s and %s" % (lib_json_loc, lib_items_json_loc))
|
|
with open(lib_json_loc, "w") as f:
|
|
json.dump(_make_lib(lib_name, lib_id, lib_create, lib_version), f, indent=2)
|
|
|
|
with open(lib_items_json_loc, "w") as f:
|
|
json.dump(_make_items(items, lib_version), f, indent=2)
|
|
|
|
|
|
def _get_item(json_object, name, value):
|
|
return [obj for obj in json_object if obj[name]==value]
|
|
|
|
|
|
def parse_options():
|
|
"""
|
|
Parse command line options
|
|
"""
|
|
parser = argparse.ArgumentParser(usage=usage())
|
|
|
|
# Run options
|
|
parser.add_argument('-n', '--name', dest='name',
|
|
help="library name")
|
|
parser.add_argument('-path', '--path', dest='path',
|
|
help="library path on storage")
|
|
parser.add_argument('--etag', dest='etag',
|
|
default='true', help="generate etag")
|
|
parser.add_argument('--skip-cert', dest='skip_cert',
|
|
default='true', help="skip OVF cert")
|
|
args = parser.parse_args()
|
|
|
|
if args.name is None or args.path is None:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
return args
|
|
|
|
def usage():
|
|
'''
|
|
The usage message for the argument parser.
|
|
'''
|
|
return """Usage: python update_library_manifests.py -n <library-name> -p <library-storage-path> --etag <true or false, default true>
|
|
--skip-cert <true or fale, default true>
|
|
|
|
"""
|
|
|
|
def main():
|
|
args = parse_options()
|
|
|
|
lib_name = args.name
|
|
lib_path = args.path
|
|
md5_enabled = args.etag == 'true' or args.etag == 'True'
|
|
skip_cert = args.skip_cert == 'true' or args.skip_cert == 'True'
|
|
|
|
make_vcsp(lib_name, lib_path, md5_enabled)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|