gcpdiag.queries.orgpolicy

29class PolicyConstraint:
30  def __init__(self, name, resource_data):
31    self.name = name
32    self._resource_data = resource_data
33
34  def __str__(self):
35    return self.name + ': ' + self._resource_data.__str__()
36
37  pass

40class BooleanPolicyConstraint(PolicyConstraint):
41  def is_enforced(self) -> bool:
42    return self._resource_data.get('enforced', False)

45class ListPolicyConstraint(PolicyConstraint):
46  def allowed_values(self) -> List[str]:
47    return self._resource_data.get('allowedValues', [])
48
49  def denied_values(self) -> List[str]:
50    return self._resource_data.get('deniedValues', [])

53class RestoreDefaultPolicyConstraint(PolicyConstraint):
54  def is_default_restored(self) -> bool:
55    """Indicates that the constraintDefault enforcement behavior is restored."""
56    return True

 74def get_effective_org_policy(project_id: str, constraint: str):
 75  """Get the effective org policy for a project and a given constraint.
 76
 77  This function will first try to get the policy from a cached list of all
 78  policies that are set on the project. If the policy is not found, it will
 79  make a direct API call to get the effective policy for the given constraint.
 80  """
 81  all_constraints = _get_effective_org_policy_all_constraints(project_id)
 82  if constraint in all_constraints:
 83    return all_constraints[constraint]
 84
 85  # If the constraint is not in the list of all policies, it means that
 86  # the policy is not set on the project. In this case, we need to get the
 87  # effective policy directly.
 88  crm_api = apis.get_api('cloudresourcemanager', 'v1', project_id)
 89  try:
 90    req = crm_api.projects().getEffectiveOrgPolicy(
 91      resource=f'projects/{project_id}', body={'constraint': constraint}
 92    )
 93    result = req.execute(num_retries=config.API_RETRIES)
 94  except googleapiclient.errors.HttpError as err:
 95    raise utils.GcpApiError(err) from err
 96
 97  if 'booleanPolicy' in result:
 98    return BooleanPolicyConstraint(result['constraint'], result['booleanPolicy'])
 99  elif 'listPolicy' in result:
100    return ListPolicyConstraint(result['constraint'], result['listPolicy'])
101  else:
102    raise ValueError(f'unknown constraint type: {result}')

105@caching.cached_api_call
106def get_all_project_org_policies(project_id: str):
107  """list all the org policies set for a particular resource.
108
109  Args:
110      project_id: The project ID.
111
112  Returns:
113      A dictionary of PolicyConstraint objects, keyed by constraint name.
114
115  Raises:
116      utils.GcpApiError: on API errors.
117  """
118  crm_api = apis.get_api('cloudresourcemanager', 'v1', project_id)
119  resource = f'projects/{project_id}'
120  all_constraints: Dict[str, PolicyConstraint] = {}
121  logging.debug('listing org policies of %s', project_id)
122
123  request = crm_api.projects().listOrgPolicies(resource=resource)
124
125  while request:
126    try:
127      response = request.execute(num_retries=config.API_RETRIES)
128    except googleapiclient.errors.HttpError as err:
129      raise utils.GcpApiError(err) from err
130
131    policies_list = response.get('policies', [])
132
133    for policy in policies_list:
134      constraint_name = policy.get('constraint')
135
136      if 'booleanPolicy' in policy:
137        all_constraints[constraint_name] = BooleanPolicyConstraint(
138          constraint_name, policy['booleanPolicy']
139        )
140      elif 'listPolicy' in policy:
141        all_constraints[constraint_name] = ListPolicyConstraint(
142          constraint_name, policy['listPolicy']
143        )
144      elif 'restoreDefault' in policy:
145        all_constraints[constraint_name] = RestoreDefaultPolicyConstraint(
146          constraint_name, policy['restoreDefault']
147        )
148      else:
149        logging.warning('unknown constraint type: %s', policy)
150
151    request = crm_api.projects().listOrgPolicies_next(request, response)
152
153  return all_constraints