summaryrefslogtreecommitdiffstats
path: root/policyhandler/pdp_api/policy_rest.py
blob: 0b33ccd5f06ebc6674e6fa8fc4245a5ba109bfd9 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
# ================================================================================
# Copyright (c) 2019-2020 AT&T Intellectual Property. All rights reserved.
# ================================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============LICENSE_END=========================================================
#

"""policy-client communicates with policy-engine thru REST API"""

import copy
import json
import os
import time
import urllib.parse
from multiprocessing.dummy import Pool as ThreadPool
from threading import Lock

import requests

from ..config import Config, Settings
from ..onap.audit import AuditHttpCode, AuditResponseCode, Metrics
from ..policy_consts import (ERRORED_POLICIES, LATEST_POLICIES, POLICY_BODY,
                             POLICY_ID, POLICY_NAMES)
from ..utils import Utils
from .pdp_consts import PDP_POLICIES, POLICY_NAME, POLICY_VERSION
from .policy_utils import PolicyUtils

_LOGGER = Utils.get_logger(__file__)

class PolicyRest(object):
    """using the http API to policy-engine"""
    PDP_API_FOLDER = os.path.basename(os.path.dirname(os.path.realpath(__file__)))
    EXPECTED_VERSIONS = "expected_versions"
    IGNORE_POLICY_NAMES = "ignore_policy_names"
    DEFAULT_TIMEOUT_IN_SECS = 60
    _lazy_inited = False

    _lock = Lock()
    _settings = Settings(Config.FIELD_POLICY_ENGINE, Config.POOL_CONNECTIONS,
                         Config.THREAD_POOL_SIZE,
                         Config.POLICY_RETRY_COUNT, Config.POLICY_RETRY_SLEEP)

    _target_entity = None
    _requests_session = None
    _url = None
    _url_pdp_decision = None
    _headers = None
    _custom_kwargs = {}
    _thread_pool_size = 4
    _policy_retry_count = 1
    _policy_retry_sleep = 0
    _timeout_in_secs = DEFAULT_TIMEOUT_IN_SECS

    @staticmethod
    def _init():
        """init static config"""
        PolicyRest._custom_kwargs = {}
        tls_ca_mode = None

        if not PolicyRest._requests_session:
            PolicyRest._requests_session = requests.Session()

        changed, pool_size = PolicyRest._settings.get_by_key(Config.POOL_CONNECTIONS, 20)
        if changed:
            PolicyRest._requests_session.mount(
                'https://', requests.adapters.HTTPAdapter(pool_connections=1,
                                                          pool_maxsize=pool_size))
            PolicyRest._requests_session.mount(
                'http://', requests.adapters.HTTPAdapter(pool_connections=1,
                                                         pool_maxsize=pool_size))

        _, config = PolicyRest._settings.get_by_key(Config.FIELD_POLICY_ENGINE)
        if config:
            PolicyRest._url = config.get("url")
            if PolicyRest._url:
                PolicyRest._url_pdp_decision = urllib.parse.urljoin(
                    PolicyRest._url, config.get("path_decision", "/decision/v1/"))
            PolicyRest._headers = config.get("headers", {})
            PolicyRest._target_entity = config.get("target_entity", Config.FIELD_POLICY_ENGINE)
            _, PolicyRest._thread_pool_size = PolicyRest._settings.get_by_key(
                Config.THREAD_POOL_SIZE, 4)
            if PolicyRest._thread_pool_size < 2:
                PolicyRest._thread_pool_size = 2

            _, PolicyRest._policy_retry_count = PolicyRest._settings.get_by_key(
                Config.POLICY_RETRY_COUNT, 1)
            _, PolicyRest._policy_retry_sleep = PolicyRest._settings.get_by_key(
                Config.POLICY_RETRY_SLEEP, 0)

            tls_ca_mode = config.get(Config.TLS_CA_MODE)
            PolicyRest._custom_kwargs = Config.get_requests_kwargs(tls_ca_mode)
            PolicyRest._timeout_in_secs = config.get(Config.TIMEOUT_IN_SECS)
            if not PolicyRest._timeout_in_secs or PolicyRest._timeout_in_secs < 1:
                PolicyRest._timeout_in_secs = PolicyRest.DEFAULT_TIMEOUT_IN_SECS

        _LOGGER.info(
            "PDP(%s) url(%s) headers(%s) tls_ca_mode(%s) timeout_in_secs(%s) custom_kwargs(%s): %s",
            PolicyRest._target_entity, PolicyRest._url_pdp_decision,
            Metrics.json_dumps(PolicyRest._headers), tls_ca_mode,
            PolicyRest._timeout_in_secs, json.dumps(PolicyRest._custom_kwargs),
            PolicyRest._settings)

        PolicyRest._settings.commit_change()
        PolicyRest._lazy_inited = True

    @staticmethod
    def reconfigure():
        """reconfigure"""
        with PolicyRest._lock:
            PolicyRest._settings.set_config(Config.discovered_config)
            if not PolicyRest._settings.is_changed():
                PolicyRest._settings.commit_change()
                return False

            PolicyRest._lazy_inited = False
            PolicyRest._init()
        return True

    @staticmethod
    def _lazy_init():
        """init static config"""
        if PolicyRest._lazy_inited:
            return

        with PolicyRest._lock:
            if PolicyRest._lazy_inited:
                return

            PolicyRest._settings.set_config(Config.discovered_config)
            PolicyRest._init()

    @staticmethod
    def _pdp_get_decision(audit, policy_ids):
        """get policies from the policy-engine by policy-ids"""
        if not PolicyRest._url:
            _LOGGER.error(
                audit.error("no url for PDP", error_code=AuditResponseCode.AVAILABILITY_ERROR))
            audit.set_http_status_code(AuditHttpCode.SERVER_INTERNAL_ERROR.value)
            return None, None

        with PolicyRest._lock:
            session = PolicyRest._requests_session
            target_entity = PolicyRest._target_entity
            url = PolicyRest._url_pdp_decision
            timeout_in_secs = PolicyRest._timeout_in_secs
            headers = copy.deepcopy(PolicyRest._headers)
            custom_kwargs = copy.deepcopy(PolicyRest._custom_kwargs)

        pdp_req = PolicyUtils.gen_req_to_pdp(policy_ids)

        metrics = Metrics(aud_parent=audit, targetEntity=target_entity, targetServiceName=url)

        headers = metrics.put_request_id_into_headers(headers)

        log_action = "post to {} at {}".format(target_entity, url)
        log_data = "msg={} headers={}, custom_kwargs({}) timeout_in_secs({})".format(
            json.dumps(pdp_req), Metrics.json_dumps(headers), json.dumps(custom_kwargs),
            timeout_in_secs)
        log_line = log_action + " " + log_data

        _LOGGER.info(metrics.metrics_start(log_line))

        res = None
        try:
            res = session.post(url, json=pdp_req, headers=headers, timeout=timeout_in_secs,
                               **custom_kwargs)
        except Exception as ex:
            error_code = (AuditHttpCode.SERVICE_UNAVAILABLE_ERROR.value
                          if isinstance(ex, requests.exceptions.RequestException)
                          else AuditHttpCode.SERVER_INTERNAL_ERROR.value)
            error_msg = ("failed {}: {} to {}".format(type(ex).__name__, str(ex), log_line))

            _LOGGER.exception(metrics.fatal(error_msg))
            metrics.set_http_status_code(error_code)
            audit.set_http_status_code(error_code)
            metrics.metrics(error_msg)
            return (error_code, None)

        log_line = "response {} from {}: text={} headers={}".format(
            res.status_code, log_line, res.text,
            Metrics.json_dumps(dict(res.request.headers.items())))

        _LOGGER.info(log_line)
        metrics.set_http_status_code(res.status_code)
        audit.set_http_status_code(res.status_code)
        metrics.metrics(log_line)

        latest_policies = None
        if res.status_code == requests.codes.ok:
            policy_bodies = res.json().get(PDP_POLICIES, {})
            latest_policies = dict((policy_id, PolicyUtils.convert_to_policy(policy))
                                   for (policy_id, policy) in policy_bodies.items())

        return res.status_code, latest_policies


    @staticmethod
    def get_latest_policy(aud_policy_id):
        """safely try retrieving the latest policy for the policy_id from the policy-engine"""
        audit, policy_id, expected_versions, ignore_policy_names = aud_policy_id
        str_metrics = "policy_id({0}), expected_versions({1}) ignore_policy_names({2})".format(
            policy_id, json.dumps(expected_versions), json.dumps(ignore_policy_names))

        try:
            return PolicyRest._get_latest_policy(
                audit, policy_id, expected_versions, ignore_policy_names, str_metrics)

        except Exception as ex:
            error_msg = ("{0}: crash {1} {2} at {3}: {4}"
                         .format(audit.request_id, type(ex).__name__, str(ex),
                                 "get_latest_policy", str_metrics))

            _LOGGER.exception(error_msg)
            audit.fatal(error_msg, error_code=AuditResponseCode.BUSINESS_PROCESS_ERROR)
            audit.set_http_status_code(AuditHttpCode.SERVER_INTERNAL_ERROR.value)
            return None


    @staticmethod
    def _get_latest_policy(audit, policy_id,
                           expected_versions, ignore_policy_names, str_metrics):
        """retry several times getting the latest policy for the policy_id from the policy-engine"""
        PolicyRest._lazy_init()
        latest_policy = None
        status_code = 0
        retry_get_config = audit.kwargs.get("retry_get_config")

        for retry in range(1, PolicyRest._policy_retry_count + 1):
            _LOGGER.debug("try(%s) retry_get_config(%s): %s", retry, retry_get_config, str_metrics)

            done, removed, latest_policy, status_code = PolicyRest._get_latest_policy_once(
                audit, policy_id, expected_versions, ignore_policy_names)

            if removed:
                audit.set_http_status_code(AuditHttpCode.DATA_NOT_FOUND_OK.value)
                return None

            if done or not retry_get_config or not PolicyRest._policy_retry_sleep:
                break

            if retry == PolicyRest._policy_retry_count:
                _LOGGER.error(
                    audit.error("gave up retrying after #{} for policy_id({}) from PDP {}"
                                .format(retry, policy_id, PolicyRest._url_pdp_decision),
                                error_code=AuditResponseCode.DATA_ERROR))
                break

            _LOGGER.warning(audit.warn(
                "will retry({}) for policy_id({}) in {} secs from PDP {}".format(
                    retry, policy_id, PolicyRest._policy_retry_sleep, PolicyRest._url_pdp_decision),
                error_code=AuditResponseCode.DATA_ERROR))
            time.sleep(PolicyRest._policy_retry_sleep)

        audit.set_http_status_code(status_code)
        if not PolicyUtils.validate_policy(latest_policy):
            audit.set_http_status_code(AuditHttpCode.DATA_NOT_FOUND_OK.value)
            _LOGGER.error(audit.error(
                "received invalid policy from PDP: {}".format(json.dumps(latest_policy)),
                error_code=AuditResponseCode.DATA_ERROR))

        return latest_policy

    @staticmethod
    def _get_latest_policy_once(audit, policy_id, expected_versions, ignore_policy_names):
        """single attempt to get the latest policy for the policy_id from the policy-engine"""

        status_code, latest_policies = PolicyRest._pdp_get_decision(audit, policy_id)

        if (ignore_policy_names and not expected_versions and not latest_policies
                and AuditHttpCode.HTTP_OK.value == status_code):
            return True, True, None, status_code

        log_line = "{} looking for policy_id({}) in latest_policies: {}".format(
            status_code, policy_id, json.dumps(latest_policies))
        _LOGGER.info(log_line)

        latest_policy = (latest_policies or {}).get(policy_id)

        log_error = ""
        if latest_policy:
            policy_body = latest_policy.get(POLICY_BODY, {})
            policy_version = policy_body.get(POLICY_VERSION)
            policy_name = policy_body.get(POLICY_NAME)
            if expected_versions and policy_version not in expected_versions:
                log_error = ("received unexpected policy version({}) instead of ({})"
                             " from PDP for policy_id={}: {}"
                             .format(policy_version, json.dumps(expected_versions),
                                     policy_id, json.dumps(latest_policy)))
            elif ignore_policy_names and policy_name in ignore_policy_names:
                log_error = ("unexpectedly received policy version({}) from PDP"
                             " for policy_id={}: {}. to ignore-policy-names {}"
                             .format(policy_version, policy_id,
                                     json.dumps(latest_policy),
                                     json.dumps(ignore_policy_names)))

        if not latest_policy or log_error:
            _LOGGER.error(audit.error(
                log_error or "received unexpected policy data({}) from PDP for policy_id={}: {}"
                .format(json.dumps(latest_policy), policy_id, json.dumps(latest_policies)),
                error_code=AuditResponseCode.DATA_ERROR))
            latest_policy = None

        done = bool(latest_policy or audit.is_serious_error(status_code))
        return done, False, latest_policy, status_code

    @staticmethod
    def get_latest_updated_policies(audit, updated_policies, removed_policies):
        """safely try retrieving the latest policies for the list of policy_names"""
        if not updated_policies and not removed_policies:
            return None, None

        policies_updated = [(policy_id, policy.get(POLICY_BODY, {}).get(POLICY_VERSION))
                            for policy_id, policy in updated_policies.items()]
        policies_removed = [(policy_id, policy.get(POLICY_NAMES, {}))
                            for policy_id, policy in removed_policies.items()]

        if not policies_updated and not policies_removed:
            return None, None

        str_metrics = "policies_updated[{0}]: {1} policies_removed[{2}]: {3}".format(
            len(policies_updated), json.dumps(policies_updated),
            len(policies_removed), json.dumps(policies_removed))

        try:
            return PolicyRest._get_latest_updated_policies(
                audit, str_metrics, policies_updated, policies_removed)

        except Exception as ex:
            error_msg = ("{0}: crash {1} {2} at {3}: {4}"
                         .format(audit.request_id, type(ex).__name__, str(ex),
                                 "get_latest_updated_policies", str_metrics))

            _LOGGER.exception(error_msg)
            audit.fatal(error_msg, error_code=AuditResponseCode.BUSINESS_PROCESS_ERROR)
            audit.set_http_status_code(AuditHttpCode.SERVER_INTERNAL_ERROR.value)
            return None, None

    @staticmethod
    def _get_latest_updated_policies(audit, str_metrics, policies_updated, policies_removed):
        """Get the latest policies of the list of policy_names from the policy-engine"""
        PolicyRest._lazy_init()
        metrics_total = Metrics(
            aud_parent=audit,
            targetEntity="{0} total get_latest_updated_policies".format(PolicyRest._target_entity),
            targetServiceName=PolicyRest._url_pdp_decision)

        metrics_total.metrics_start("get_latest_updated_policies {0}".format(str_metrics))
        _LOGGER.debug(str_metrics)

        policies_to_find = {}
        for (policy_id, policy_version) in policies_updated:
            if not policy_id or policy_version is None:
                continue
            policy = policies_to_find.get(policy_id)
            if not policy:
                policies_to_find[policy_id] = {
                    POLICY_ID: policy_id,
                    PolicyRest.EXPECTED_VERSIONS: {policy_version: True},
                    PolicyRest.IGNORE_POLICY_NAMES: {}
                }
                continue
            policy[PolicyRest.EXPECTED_VERSIONS][policy_version] = True

        for (policy_id, policy_names) in policies_removed:
            if not policy_id:
                continue
            policy = policies_to_find.get(policy_id)
            if not policy:
                policies_to_find[policy_id] = {
                    POLICY_ID: policy_id,
                    PolicyRest.IGNORE_POLICY_NAMES: policy_names
                }
                continue
            policy[PolicyRest.IGNORE_POLICY_NAMES].update(policy_names)

        apns = [(audit, policy_id,
                 policy_to_find.get(PolicyRest.EXPECTED_VERSIONS),
                 policy_to_find.get(PolicyRest.IGNORE_POLICY_NAMES))
                for (policy_id, policy_to_find) in policies_to_find.items()]

        policies = None
        apns_length = len(apns)
        _LOGGER.debug("apns_length(%s) policies_to_find %s", apns_length,
                      json.dumps(policies_to_find))

        if apns_length == 1:
            policies = [PolicyRest.get_latest_policy(apns[0])]
        else:
            pool = ThreadPool(min(PolicyRest._thread_pool_size, apns_length))
            policies = pool.map(PolicyRest.get_latest_policy, apns)
            pool.close()
            pool.join()

        metrics_total.metrics("result({}) get_latest_updated_policies {}: {} {}"
                              .format(apns_length, str_metrics,
                                      len(policies), json.dumps(policies)))

        updated_policies = dict((policy[POLICY_ID], policy)
                                for policy in policies
                                if policy and policy.get(POLICY_ID))

        removed_policies = dict((policy_id, True)
                                for (policy_id, policy_to_find) in policies_to_find.items()
                                if not policy_to_find.get(PolicyRest.EXPECTED_VERSIONS)
                                and policy_to_find.get(PolicyRest.IGNORE_POLICY_NAMES)
                                and policy_id not in updated_policies)

        errored_policies = dict((policy_id, policy_to_find)
                                for (policy_id, policy_to_find) in policies_to_find.items()
                                if policy_id not in updated_policies
                                and policy_id not in removed_policies)

        _LOGGER.debug(
            "result(%s) updated_policies %s, removed_policies %s, errored_policies %s",
            apns_length, json.dumps(updated_policies), json.dumps(removed_policies),
            json.dumps(errored_policies))

        if errored_policies:
            audit.set_http_status_code(AuditHttpCode.DATA_ERROR.value)
            audit.error(
                "errored_policies in PDP: {}".format(json.dumps(errored_policies)),
                error_code=AuditResponseCode.DATA_ERROR)

        return updated_policies, removed_policies


    @staticmethod
    def get_latest_policies(audit, policy_ids=None):
        """Get the latest policies by policy-ids from the policy-engine"""
        result = {}
        str_policy_ids = json.dumps(policy_ids or [])

        try:
            PolicyRest._lazy_init()
            if policy_ids:
                _, latest_policies = PolicyRest._pdp_get_decision(audit, policy_ids)

                if not latest_policies:
                    audit.set_http_status_code(AuditHttpCode.DATA_NOT_FOUND_OK.value)
                    _LOGGER.warning(audit.warn(
                        "received no policies from PDP for policy_ids {}"
                        .format(str_policy_ids), error_code=AuditResponseCode.DATA_ERROR))
                    return latest_policies

                valid_policies = {}
                errored_policies = {}
                for (policy_id, policy) in latest_policies.items():
                    if PolicyUtils.validate_policy(policy):
                        valid_policies[policy_id] = policy
                    else:
                        errored_policies[policy_id] = policy

                result[LATEST_POLICIES] = valid_policies
                result[ERRORED_POLICIES] = errored_policies

            _LOGGER.debug("got policies for policy_ids: %s. result: %s",
                          str_policy_ids, json.dumps(result))
            return result

        except Exception as ex:
            error_msg = ("{}: crash {} {} at {}: {}"
                         .format(audit.request_id, type(ex).__name__, str(ex),
                                 "get_latest_policies", str_policy_ids))

            _LOGGER.exception(error_msg)
            audit.fatal(error_msg, error_code=AuditResponseCode.BUSINESS_PROCESS_ERROR)
            audit.set_http_status_code(AuditHttpCode.SERVER_INTERNAL_ERROR.value)
            return None