1# Copyright 2020 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#      http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15import json
16import mock
17import os
18import time
19from os import path
20
21
22import google.auth
23import google.auth.credentials
24from google.auth import environment_vars
25from google.auth.transport import mtls
26import google.auth.transport.requests
27import google.auth.transport.urllib3
28
29MTLS_ENDPOINT = "https://pubsub.mtls.googleapis.com/v1/projects/{}/topics"
30REGULAR_ENDPOINT = "https://pubsub.googleapis.com/v1/projects/{}/topics"
31
32
33def test_requests():
34    credentials, project_id = google.auth.default()
35    credentials = google.auth.credentials.with_scopes_if_required(
36        credentials, ["https://www.googleapis.com/auth/pubsub"]
37    )
38
39    authed_session = google.auth.transport.requests.AuthorizedSession(credentials)
40    with mock.patch.dict(os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}):
41        authed_session.configure_mtls_channel()
42
43    # If the devices has default client cert source, then a mutual TLS channel
44    # is supposed to be created.
45    assert authed_session.is_mtls == mtls.has_default_client_cert_source()
46
47    # Sleep 1 second to avoid 503 error.
48    time.sleep(1)
49
50    if authed_session.is_mtls:
51        response = authed_session.get(MTLS_ENDPOINT.format(project_id))
52    else:
53        response = authed_session.get(REGULAR_ENDPOINT.format(project_id))
54
55    assert response.ok
56
57
58def test_urllib3():
59    credentials, project_id = google.auth.default()
60    credentials = google.auth.credentials.with_scopes_if_required(
61        credentials, ["https://www.googleapis.com/auth/pubsub"]
62    )
63
64    authed_http = google.auth.transport.urllib3.AuthorizedHttp(credentials)
65    with mock.patch.dict(os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}):
66        is_mtls = authed_http.configure_mtls_channel()
67
68    # If the devices has default client cert source, then a mutual TLS channel
69    # is supposed to be created.
70    assert is_mtls == mtls.has_default_client_cert_source()
71
72    # Sleep 1 second to avoid 503 error.
73    time.sleep(1)
74
75    if is_mtls:
76        response = authed_http.request("GET", MTLS_ENDPOINT.format(project_id))
77    else:
78        response = authed_http.request("GET", REGULAR_ENDPOINT.format(project_id))
79
80    assert response.status == 200
81
82
83def test_requests_with_default_client_cert_source():
84    credentials, project_id = google.auth.default()
85    credentials = google.auth.credentials.with_scopes_if_required(
86        credentials, ["https://www.googleapis.com/auth/pubsub"]
87    )
88
89    authed_session = google.auth.transport.requests.AuthorizedSession(credentials)
90
91    if mtls.has_default_client_cert_source():
92        with mock.patch.dict(os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}):
93            authed_session.configure_mtls_channel(
94                client_cert_callback=mtls.default_client_cert_source()
95            )
96
97        assert authed_session.is_mtls
98
99        # Sleep 1 second to avoid 503 error.
100        time.sleep(1)
101
102        response = authed_session.get(MTLS_ENDPOINT.format(project_id))
103        assert response.ok
104
105
106def test_urllib3_with_default_client_cert_source():
107    credentials, project_id = google.auth.default()
108    credentials = google.auth.credentials.with_scopes_if_required(
109        credentials, ["https://www.googleapis.com/auth/pubsub"]
110    )
111
112    authed_http = google.auth.transport.urllib3.AuthorizedHttp(credentials)
113
114    if mtls.has_default_client_cert_source():
115        with mock.patch.dict(os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}):
116            assert authed_http.configure_mtls_channel(
117                client_cert_callback=mtls.default_client_cert_source()
118            )
119
120        # Sleep 1 second to avoid 503 error.
121        time.sleep(1)
122
123        response = authed_http.request("GET", MTLS_ENDPOINT.format(project_id))
124        assert response.status == 200
125