validators/src/HostOSValidation.py

   1 #!/usr/bin/python
   2 # Copyright 2019 Nokia
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #    http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 import logging
  17 import json
  18 import re
  19
  20 from cmframework.apis import cmvalidator
  21 from cmdatahandlers.api import validation
  22
  23
  24 class HostOSValidation(cmvalidator.CMValidator):
  25     domain = 'cloud.host_os'
  26     GRUB2_PASSWORD_PATTERN = r"^grub\.pbkdf2\.sha512\.\d+\.[0-9A-F]+\.[0-9A-F]+$"
  27
  28     def get_subscription_info(self):
  29         logging.debug('get_subscription info called')
  30         return r'^cloud\.host_os$'
  31
  32     def validate_set(self, dict_key_value):
  33         grub2pass_attr = 'grub2_password'
  34         lockout_time_attr = 'lockout_time'
  35         failed_login_attempts_attr = 'failed_login_attempts'
  36         logging.debug('validate_set called with %s' % str(dict_key_value))
  37
  38         value_str = dict_key_value.get(self.domain, None)
  39         logging.debug('{0} domain value: {1}'.format(self.domain, value_str))
  40         if value_str is not None:
  41             value_dict = json.loads(value_str)
  42
  43             if not isinstance(value_dict, dict):
  44                 raise validation.ValidationError('%s value is not a dict' % self.domain)
  45
  46             passwd = value_dict.get(grub2pass_attr)
  47             if passwd:
  48                 self.validate_passwd_hash(passwd)
  49
  50             lockout_t = value_dict.get(lockout_time_attr)
  51             if lockout_t:
  52                 self.validate_lockout_time(lockout_t)
  53
  54             failed_login_a = value_dict.get(failed_login_attempts_attr)
  55             if failed_login_a:
  56                 self.validate_failed_login_attempts(failed_login_a)
  57         else:
  58             raise validation.ValidationError('Missing domain: %s' % self.domain)
  59
  60     def validate_delete(self, dict_key_value):
  61         logging.debug('validate_delete called with %s' % str(dict_key_value))
  62         raise validation.ValidationError('%s cannot be deleted' % self.domain)
  63
  64     def validate_passwd_hash(self, passwd_hash):
  65         if not re.match(self.GRUB2_PASSWORD_PATTERN, passwd_hash):
  66             raise validation.ValidationError('The passwd hash: "%s" is not a valid hash!' % passwd_hash)
  67
  68     def validate_lockout_time(self, _lockout_time):
  69         if not re.match(r"^[0-9]+$", str(_lockout_time)):
  70             raise validation.ValidationError('The lockout time: "%s" is not valid!' % _lockout_time)
  71
  72     def validate_failed_login_attempts(self, _failed_login_attempts):
  73         if not re.match(r"^[0-9]+$", str(_failed_login_attempts)):
  74             raise validation.ValidationError('The failed login attempts: "%s" is not valid!' % _failed_login_attempts)