ucloud-api/schemas.py

import json

from common_fields import Field, VmUUIDField, SpecsField
from helper import check_otp
from etcd3_wrapper import Etcd3Wrapper

client = Etcd3Wrapper()


class BaseSchema(object):
    def __init__(self, data, fields=None):
        _ = data

        self.__errors = []
        if fields is None:
            self.fields = []
        else:
            self.fields = fields

    def validation(self):
        # custom validation is optional
        return True

    def is_valid(self):
        for field in self.fields:
            field.is_valid()
            self.add_field_errors(field)

        for parent in self.__class__.__bases__:
            try:
                parent.validation(self)
            except AttributeError:
                pass
        if not self.__errors:
            self.validation()

        if self.__errors:
            return False
        return True

    def get_errors(self):
        return {"Message": self.__errors}

    def add_field_errors(self, field: Field):
        self.__errors += field.get_errors()

    def add_error(self, error):
        self.__errors.append(error)


class OTPSchema(BaseSchema):
    def __init__(self, data: dict, fields=None):
        self.name = Field("name", str, data.get("name", KeyError))
        self.realm = Field("realm", str, data.get("realm", KeyError))
        self.token = Field("token", str, data.get("token", KeyError))

        _fields = [self.name, self.realm, self.token]
        if fields:
            _fields += fields
        super().__init__(data=data, fields=_fields)

    def validation(self):
        rc = check_otp(self.name.value, self.realm.value, self.token.value)
        if rc != 200:
            self.add_error("Wrong Credentials")


class CreateVMSchema(OTPSchema):
    def __init__(self, data):
        self.specs = SpecsField(data)

        self.image_uuid = Field("image_uuid", str, data.get("image_uuid", KeyError))
        self.image_uuid.validation = self.image_uuid_validation

        fields = [self.specs, self.image_uuid]
        super().__init__(data=data, fields=fields)

    def image_uuid_validation(self):
        images = client.get_prefix("/v1/image/")

        if self.image_uuid.value not in [i.key.split("/")[-1] for i in images]:
            self.add_error("Image UUID not valid")


class VMStatusSchema(BaseSchema):
    def __init__(self, data):
        self.uuid = VmUUIDField(data)

        fields = [self.uuid]

        super().__init__(data, fields)


class CreateImageSchema(BaseSchema):
    def __init__(self, data):
        # Fields
        self.uuid: Field = Field("uuid", str, data.get("uuid", KeyError))
        self.name = Field("name", str, data.get("name", KeyError))
        self.image_store = Field("image_store", str, data.get("image_store", KeyError))
        # Validations
        self.uuid.validation = self.file_uuid_validation

        # All Fields
        fields = [self.uuid, self.name, self.image_store]
        super().__init__(data, fields)

    def file_uuid_validation(self):
        file_entry = client.get(f"/v1/file/{self.uuid.value}")
        if file_entry is None:
            self.add_error(f"Image File with uuid '{self.uuid.value}' Not Found")

    def image_store_name_validation(self):
        image_stores = list(client.get_prefix("/v1/image_store/"))

        image_store = next(filter(lambda s: json.loads(s.value)["name"] == self.image_store.value,
                                  image_stores))
        if not image_store:
            self.add_error(f"Store '{self.image_store.value}' does not exists")


class VmActionSchema(OTPSchema):
    def __init__(self, data):
        self.uuid = VmUUIDField(data)
        self.action = Field("action", str, data.get("action", KeyError))

        self.action.validation = self.action_validation

        fields = [self.uuid, self.action]
        super().__init__(data=data, fields=fields)

    def action_validation(self):
        allowed_actions = ["start", "shutdown", "suspend", "resume", "delete"]
        if self.action.value not in allowed_actions:
            self.add_error(f"Invalid Action. Allowed Actions are {allowed_actions}")

    def validation(self):
        vm = client.get(f"/v1/vm/{self.uuid.value}", value_in_json=True)
        if vm.value["owner"] != self.name:
            self.add_error("Invalid User")


class CreateHostSchema(OTPSchema):
    def __init__(self, data):
        self.specs = SpecsField(data)
        self.hostname = Field("hostname", str, data.get("hostname", KeyError))

        fields = [self.specs, self.hostname]

        super().__init__(data=data, fields=fields)

    def validation(self):
        if self.realm.value != "ungleich-admin":
            self.add_error("Invalid Credentials/Insufficient Permission")