mirror of
https://github.com/goharbor/harbor.git
synced 2024-11-30 06:03:45 +01:00
118 lines
4.5 KiB
Python
118 lines
4.5 KiB
Python
import os
|
|
import string
|
|
import random
|
|
|
|
from g import DEFAULT_UID, DEFAULT_GID
|
|
from pathlib import Path
|
|
|
|
# To meet security requirement
|
|
# By default it will change file mode to 0600, and make the owner of the file to 10000:10000
|
|
def mark_file(path, mode=0o600, uid=DEFAULT_UID, gid=DEFAULT_GID):
|
|
if mode > 0:
|
|
os.chmod(path, mode)
|
|
if uid > 0 and gid > 0:
|
|
os.chown(path, uid, gid)
|
|
|
|
|
|
def validate(conf, **kwargs):
|
|
# Protocol validate
|
|
protocol = conf.get("configuration", "ui_url_protocol")
|
|
if protocol != "https" and kwargs.get('notary_mode'):
|
|
raise Exception(
|
|
"Error: the protocol must be https when Harbor is deployed with Notary")
|
|
if protocol == "https":
|
|
if not conf.has_option("configuration", "ssl_cert"):
|
|
raise Exception(
|
|
"Error: The protocol is https but attribute ssl_cert is not set")
|
|
cert_path = conf.get("configuration", "ssl_cert")
|
|
if not os.path.isfile(cert_path):
|
|
raise Exception(
|
|
"Error: The path for certificate: %s is invalid" % cert_path)
|
|
if not conf.has_option("configuration", "ssl_cert_key"):
|
|
raise Exception(
|
|
"Error: The protocol is https but attribute ssl_cert_key is not set")
|
|
cert_key_path = conf.get("configuration", "ssl_cert_key")
|
|
if not os.path.isfile(cert_key_path):
|
|
raise Exception(
|
|
"Error: The path for certificate key: %s is invalid" % cert_key_path)
|
|
|
|
# Storage validate
|
|
valid_storage_drivers = ["filesystem",
|
|
"azure", "gcs", "s3", "swift", "oss"]
|
|
storage_provider_name = conf.get(
|
|
"configuration", "registry_storage_provider_name").strip()
|
|
if storage_provider_name not in valid_storage_drivers:
|
|
raise Exception("Error: storage driver %s is not supported, only the following ones are supported: %s" % (
|
|
storage_provider_name, ",".join(valid_storage_drivers)))
|
|
|
|
storage_provider_config = conf.get(
|
|
"configuration", "registry_storage_provider_config").strip()
|
|
if storage_provider_name != "filesystem":
|
|
if storage_provider_config == "":
|
|
raise Exception(
|
|
"Error: no provider configurations are provided for provider %s" % storage_provider_name)
|
|
|
|
# Redis validate
|
|
redis_host = conf.get("configuration", "redis_host")
|
|
if redis_host is None or len(redis_host) < 1:
|
|
raise Exception(
|
|
"Error: redis_host in harbor.cfg needs to point to an endpoint of Redis server or cluster.")
|
|
|
|
redis_port = conf.get("configuration", "redis_port")
|
|
if len(redis_port) < 1:
|
|
raise Exception(
|
|
"Error: redis_port in harbor.cfg needs to point to the port of Redis server or cluster.")
|
|
|
|
redis_db_index = conf.get("configuration", "redis_db_index").strip()
|
|
if len(redis_db_index.split(",")) != 3:
|
|
raise Exception(
|
|
"Error invalid value for redis_db_index: %s. please set it as 1,2,3" % redis_db_index)
|
|
|
|
def validate_crt_subj(dirty_subj):
|
|
subj_list = [item for item in dirty_subj.strip().split("/") \
|
|
if len(item.split("=")) == 2 and len(item.split("=")[1]) > 0]
|
|
return "/" + "/".join(subj_list)
|
|
|
|
|
|
def generate_random_string(length):
|
|
return ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(length))
|
|
|
|
|
|
def prepare_config_dir(root, *name):
|
|
absolute_path = os.path.join(root, *name)
|
|
if not os.path.exists(absolute_path):
|
|
os.makedirs(absolute_path)
|
|
return absolute_path
|
|
|
|
def prepare_dir(root: str, *args, **kwargs) -> str:
|
|
gid, uid = kwargs.get('gid'), kwargs.get('uid')
|
|
absolute_path = Path(os.path.join(root, *args))
|
|
if absolute_path.is_file():
|
|
raise Exception('Path exists and the type is regular file')
|
|
mode = kwargs.get('mode') or 0o755
|
|
absolute_path.mkdir(mode, parents=True, exist_ok=True)
|
|
|
|
# if uid or gid not None, then change the ownership of this dir
|
|
if not(gid is None and uid is None):
|
|
dir_uid, dir_gid = absolute_path.stat().st_uid, absolute_path.stat().st_gid
|
|
if uid is None:
|
|
uid = dir_uid
|
|
if gid is None:
|
|
gid = dir_gid
|
|
os.chown(absolute_path, uid, gid)
|
|
|
|
return str(absolute_path)
|
|
|
|
|
|
|
|
def delfile(src):
|
|
if os.path.isfile(src):
|
|
try:
|
|
os.remove(src)
|
|
print("Clearing the configuration file: %s" % src)
|
|
except Exception as e:
|
|
print(e)
|
|
elif os.path.isdir(src):
|
|
for item in os.listdir(src):
|
|
itemsrc = os.path.join(src, item)
|
|
delfile(itemsrc) |