controller/rootfs/scheduler/resources/secret.py at cf4cf4505ef7fb83de2dc2f84bb6bd7c8fa0e655 · drycc/controller · GitHub

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
import base64
import json
from scheduler.resources import Resource
from scheduler.exceptions import KubeHTTPException, KubeException


class Secret(Resource):
    def get(self, namespace, name=None, **kwargs):
        """
        Fetch a single Secret or a list
        """
        url = '/namespaces/{}/secrets'
        args = [namespace]
        if name is not None:
            args.append(name)
            url += '/{}'
            message = 'get Secret "{}" in Namespace "{}"'
        else:
            message = 'get Secrets in Namespace "{}"'

        url = self.api(url, *args)
        response = self.http_get(url, params=self.query_params(**kwargs))
        if self.unhealthy(response.status_code):
            args.reverse()  # error msg is in reverse order
            raise KubeHTTPException(response, message, *args)

        # return right away if it is a list
        if name is None:
            return response

        # decode the base64 data
        secrets = response.json()
        for key, value in secrets['data'].items():
            if value is None:
                secrets['data'][key] = ''
                continue

            value = base64.b64decode(value)
            value = value if isinstance(value, bytes) else bytes(str(value), 'UTF-8')
            secrets['data'][key] = value.decode(encoding='UTF-8')

        # tell python-requests it actually hasn't consumed the data
        response._content = bytes(json.dumps(secrets), 'UTF-8')

        return response

    def manifest(self, namespace, name, data, secret_type='Opaque', labels={}):
        secret_types = ['Opaque', 'kubernetes.io/dockerconfigjson']
        if secret_type not in secret_types:
            raise KubeException('{} is not a supported secret type. Use one of the following: '.format(secret_type, ', '.join(secret_types)))  # noqa

        manifest = {
            'kind': 'Secret',
            'apiVersion': 'v1',
            'metadata': {
                'name': name,
                'namespace': namespace,
                'labels': {
                    'app': namespace,
                    'heritage': 'deis'
                }
            },
            'type': secret_type,
            'data': {}
        }

        # add in any additional label info
        manifest['metadata']['labels'].update(labels)

        for key, value in data.items():
            if value is None:
                manifest['data'].update({key: ''})
                continue

            value = value if isinstance(value, bytes) else bytes(str(value), 'UTF-8')
            item = base64.b64encode(value).decode(encoding='UTF-8')
            manifest['data'].update({key: item})

        return manifest

    def create(self, namespace, name, data, secret_type='Opaque', labels={}):
        manifest = self.manifest(namespace, name, data, secret_type, labels)
        url = self.api("/namespaces/{}/secrets", namespace)
        response = self.http_post(url, json=manifest)
        if self.unhealthy(response.status_code):
            raise KubeHTTPException(
                response,
                'failed to create Secret "{}" in Namespace "{}"', name, namespace
            )

        return response

    def update(self, namespace, name, data, secret_type='Opaque', labels={}):
        manifest = self.manifest(namespace, name, data, secret_type, labels)
        url = self.api("/namespaces/{}/secrets/{}", namespace, name)
        response = self.http_put(url, json=manifest)
        if self.unhealthy(response.status_code):
            raise KubeHTTPException(
                response,
                'failed to update Secret "{}" in Namespace "{}"',
                name, namespace
            )

        return response

    def delete(self, namespace, name):
        url = self.api("/namespaces/{}/secrets/{}", namespace, name)
        response = self.http_delete(url)
        if self.unhealthy(response.status_code):
            raise KubeHTTPException(
                response,
                'delete Secret "{}" in Namespace "{}"', name, namespace
            )

        return response