File manager - Edit - /home/u478019808/domains/bestandroidphones.store/public_html/static/img/logo/system_tests.tar
Back
system_tests_sync/test_oauth2_credentials.py 0000644 00000003771 15025175721 0015553 0 ustar 00 # Copyright 2016 Google LLC # # 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. import json from google.auth import _helpers import google.oauth2.credentials GOOGLE_OAUTH2_TOKEN_ENDPOINT = "https://oauth2.googleapis.com/token" def test_refresh(authorized_user_file, http_request, token_info): with open(authorized_user_file, "r") as fh: info = json.load(fh) credentials = google.oauth2.credentials.Credentials( None, # No access token, must be refreshed. refresh_token=info["refresh_token"], token_uri=GOOGLE_OAUTH2_TOKEN_ENDPOINT, client_id=info["client_id"], client_secret=info["client_secret"], ) credentials.refresh(http_request) assert credentials.token info = token_info(credentials.token) info_scopes = _helpers.string_to_scopes(info["scope"]) # Canonical list of scopes at https://cloud.google.com/sdk/gcloud/reference/auth/application-default/login # or do `gcloud auth application-defaut login --help` canonical_scopes = set( [ "https://www.googleapis.com/auth/userinfo.email", "https://www.googleapis.com/auth/cloud-platform", "openid", ] ) # When running the test locally, we always have an additional "accounts.reauth" scope. canonical_scopes_with_reauth = canonical_scopes.copy() canonical_scopes_with_reauth.add("https://www.googleapis.com/auth/accounts.reauth") assert set(info_scopes) == canonical_scopes or set(info_scopes) == canonical_scopes_with_reauth system_tests_sync/test_mtls_http.py 0000644 00000010510 15025175721 0013777 0 ustar 00 # Copyright 2020 Google LLC # # 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. import json import mock import os import time from os import path import google.auth import google.auth.credentials from google.auth import environment_vars from google.auth.transport import mtls import google.auth.transport.requests import google.auth.transport.urllib3 MTLS_ENDPOINT = "https://pubsub.mtls.googleapis.com/v1/projects/{}/topics" REGULAR_ENDPOINT = "https://pubsub.googleapis.com/v1/projects/{}/topics" def test_requests(): credentials, project_id = google.auth.default() credentials = google.auth.credentials.with_scopes_if_required( credentials, ["https://www.googleapis.com/auth/pubsub"] ) authed_session = google.auth.transport.requests.AuthorizedSession(credentials) with mock.patch.dict(os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}): authed_session.configure_mtls_channel() # If the devices has default client cert source, then a mutual TLS channel # is supposed to be created. assert authed_session.is_mtls == mtls.has_default_client_cert_source() # Sleep 1 second to avoid 503 error. time.sleep(1) if authed_session.is_mtls: response = authed_session.get(MTLS_ENDPOINT.format(project_id)) else: response = authed_session.get(REGULAR_ENDPOINT.format(project_id)) assert response.ok def test_urllib3(): credentials, project_id = google.auth.default() credentials = google.auth.credentials.with_scopes_if_required( credentials, ["https://www.googleapis.com/auth/pubsub"] ) authed_http = google.auth.transport.urllib3.AuthorizedHttp(credentials) with mock.patch.dict(os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}): is_mtls = authed_http.configure_mtls_channel() # If the devices has default client cert source, then a mutual TLS channel # is supposed to be created. assert is_mtls == mtls.has_default_client_cert_source() # Sleep 1 second to avoid 503 error. time.sleep(1) if is_mtls: response = authed_http.request("GET", MTLS_ENDPOINT.format(project_id)) else: response = authed_http.request("GET", REGULAR_ENDPOINT.format(project_id)) assert response.status == 200 def test_requests_with_default_client_cert_source(): credentials, project_id = google.auth.default() credentials = google.auth.credentials.with_scopes_if_required( credentials, ["https://www.googleapis.com/auth/pubsub"] ) authed_session = google.auth.transport.requests.AuthorizedSession(credentials) if mtls.has_default_client_cert_source(): with mock.patch.dict(os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}): authed_session.configure_mtls_channel( client_cert_callback=mtls.default_client_cert_source() ) assert authed_session.is_mtls # Sleep 1 second to avoid 503 error. time.sleep(1) response = authed_session.get(MTLS_ENDPOINT.format(project_id)) assert response.ok def test_urllib3_with_default_client_cert_source(): credentials, project_id = google.auth.default() credentials = google.auth.credentials.with_scopes_if_required( credentials, ["https://www.googleapis.com/auth/pubsub"] ) authed_http = google.auth.transport.urllib3.AuthorizedHttp(credentials) if mtls.has_default_client_cert_source(): with mock.patch.dict(os.environ, {environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE: "true"}): assert authed_http.configure_mtls_channel( client_cert_callback=mtls.default_client_cert_source() ) # Sleep 1 second to avoid 503 error. time.sleep(1) response = authed_http.request("GET", MTLS_ENDPOINT.format(project_id)) assert response.status == 200 system_tests_sync/secrets.tar.enc 0000644 00000024123 15025175721 0013301 0 ustar 00 $ �MU Udg���`* �� �+%�Xf�'�� >�4I�P �eU-��uS$�MW=_P�\X?�f�X����ȳ������_u������5�C�����>p#q��8P����wG�|�5X9�T.6(,?���0� K��ewM@�,WpmFy<���H���,b��RW#�� ��MN ��\��Ej���#&����RUzi��% ����a �h�C8��>?b?Ԏ����]�n�W;���H1� ��/����t뎩�W������2��Bž�K2� �� a���gbW2/���Α ��-�(1� ��@�+(��T��I�'E��Uו���f��w� � ����9,��{%�!�jCM�F�w�j�ȃ���a��N-�Q`aҔ��Y:��s� E ��c4�^�{�*G�f$�z<. sb;j�gh�z YI�ϰ��w�P{:���|�����5�g��az�����W��"�) �e%�g�� )��?�4 Q:�at��{��e,aS�+`���]�����s�g�Aܲ���>�A�rT����� c�L��u�\=kE��v����sf�V3��k8؇ҸL} ���� ػ8G(*Qj l�4��p��B�զ�=X���J�������# ;ޤ�c9������E�~�ftГnhu��Uw bv`/�q�h��.�h>�P�Qn��PэԐ����\`)|b���<$����y�8腼 g����<�<n_��i� T�q��s�� rf?�y�@��$j��@ǩ�#פ !���ݵ�e �?�V_�� �������[٤(W��m��Ŀ��_9(�ma�rVHG<��ΞP�l�z��1h���| e+i_F�m|c�A=o8x�@6��z�[O�xO��v�*�Դ��; w9�tőa�n�YX�0z�pL�t ��P��j�R��D�� ^.��� OsC��;�&�E�]n�/!~C��J�Lʪ�T � ����|��y��q `A�,��� ٮ�`�Z��T��������GqO����=�ء�3hj��AM�@p�Y��� �y,g0�C�?��p�6��.��� ���� ��k�a6����ł>pn�����k ��@��w�o.'�:9�i+y^T�� Q��ޢ�R'%H��ܠ��?����K�,�P����7|�i�d"V5L� �B`�8Q��Q�o8��O�(���zxI��BR$��|9Q��a'���ߢ� �'Eh[�B��*^f VL�U��B�Vܤ��FU+��b�������/c�)')ë�8�Y�����2H����C��f�!$�U�5��DM�(]B=T(+f�u�s��&���7 A�XX�wW��Ѹ�� �Q[�ΰ�+@#b���yw ֟#��g��{���#?�9�v�P�oA�ճ-���* 9EQo���k��1��� hq�`�wmJn���u�s��-�~H,��_�Ox38c�6�j����L�d^yx[i3��W�L�~�n�D�vY�N)w��J�D� b(#�ص��r��P�?���R�OHF�U>��SʐJ)�I�#��#��v'�}� C�I�nY� �da��$ I<�����P� ���� b� �Y��W��q(��7�O乂��ڃ�)��\Q��3S��S� �V�`6�e�9���@*q��nT��T�������8��[Y�9T�A�v��Y��WԨi'��.�m(���� �k9?g��� u#� �z��ݭp%!S_{�ƺ�Yo�c�sA# &�|ƥ� �$`y!�ާ� 86�^ ��Zj��L��?����Sm�<�g[���| ƜH��s4 At�0#�J��VC6�5�|���0�쇟�s%���s���d:�f&c+r�r�pBg]Q� ���p�:��&[E��.�JN���N����ކ^y��̊A�r� (���R�B�&��dXi��}��w��ǵ�D������n��,��]����B��~�q� 7ݗ Eր�Eo�:z�Ģ͂��Vԧ����mU����q��.�pڏV�yP�FB�S�]� v��� �( �;����+;��;�� g %6F�6[֗��I����#. �� ���{�p`��,i��r�k��.Ԍ�߿z#-q%��'�o-�V�l�6����}2���ט������&��s^a�|��&٩� ��Lo�oW���>�Y�= �RqG��0�@N�v� w��C8����U;o�g;W���K � �lPYo�C��9�c9�\d� O���`�En�m9 �n�%�� �eG��m���k��:���4��%���x��HB *��}x�����Iv �:�Z�-�M/� ��E��h��b ��d��lDQ�dIg�K!&S�3���ܼ�nZ#I�̻��T�q1���׀~�z�T�X&� , �~�C{�5ٯ !/7�ފn���mn�s����a�fj�Ώ��Q��=7[����SI��kУ�����^~S y�A�؉Y^�hhECN�c�� `�f .��� [\�ѸY�C��t�6m\52�kJHd`v�?��.E d���=0��S�~�` /�1��PL�H��[�{:�C@�� ���HReߧ�}H3� ��� ��x�H\���Mbx�i��ֲ�]?{��A��8�/s�P�R!aß{� F��WV�q�D\e����.)�l�#��ك�8�I3s~ �|�i6@B��DtA��&��|�.Ipf��a�u��$�u� �m���E�v4J& �g8HD���y'�ݮ��鹕��7�u�a��Tw@p� �y(�0 ���L��� �C��Y �9�VE�$�_�z_"�'P��d��m��K��������o�O�\��'lmdJ�r�T�nB�I��� %�;#��] �g ��} �Z&g�4��~̈U�� �v����zq��1�Gi����jVuٙ�{�^�f�t�]�P�Lp2l� Pm8���=�[ �7�ї^��ǻ�c݆� �4ݩ�Q @��}z����!s��^H�tB��W�U���-��� ��}R? Ք��p<9ߕJ� . �R����LD ��9]��e���"�+) 2 �vE�)A�&�ZsMc(��Hʙ�Ქ`�z�/_��l�z� �\�/� p�G�{�>)$V) xV]��p� g.jTN-�'�$�Bf�gN=p�פJ�b����P��%:������g��𝨍t� �E>�rؿ��T�3mpJi�a-���V�ݎMk:�&���������趽,�t&e�Anfc�����;�HL�u���+r ז7�Q��z �օ-��E�����sn�έ��InLގ?�_��<7'n��9ś %� n�cgl���zFQ;c�`f5-��G���9�3�v� �g+��?��m �T"��4��)Z�brd��� ��2C~��༹Q ���V��~|S��7&�E۹Q�zm��:�T��YV1�6"���p�>44E���R+�h�?�@�� �%1?� �5�mp/�6���_�p D� ��,x�U�ˎ#�a���%�Ͽ\u�6JI�����F��n��Oh9� ؖc�i��I.�M>��u)Y\��e-�ώ��|~]ٗ�}�ɇ ����P������/� GCQԸ���ي$��O�f9NŃ� #$Ҧ�;�3�Җahs�{�� �k�a��y�%ʯ�r��ł��3g �E�G4^0�Dqc�>���8e0�]��Sf�|�i���M�v�c>��/f�M�5��y�l��Y��~����G!udO�(�U�� � g��-��K3���@��aLVh�~�5�wI{d��r���[�o�^E���w���| =�vWQ)6��糗�� ��^l%�)�d�|yZ ����"�/� �-�f�6��m��a"�eyFU�ܐ�_��Uot9���_9!X� �B\#�����U�U �ٹ�'�#3ec1+ҵh�;���yu �0^����`X�2��-�eA��U��ݐ�qv���-�k���<��[Q�� L!�#�f�1�F��k%���m���V.64X��Z�a��A羐��β���nj�63��ݛ�V:o_k���=�F_`�Û�bԴ�F�rYb��;���gL�*8r����hX�U=?\/<sP���>/HC��@���( ������=:ߚKi��}�_O���jH���y��/'�)�1�L1���<� )s������ |W :x �h�&Pk>� n�WR E�S�[��s�iɄ�A9a���ɓۤ�AŌ�4���[- P�Bf O������!p>�[�;�~������dU� �����|5�y�w\�� �:� ʺx�O �U�%�@L�Icn��*V�i�:�Y upq<S�B-���d�غlU)$���f�2�S,W�QR�L�$�V;'&�2�X�t �8K�a�_6C�Z�{�I�I�����g^�u{l�:���5�rj�%�� �";Q�L��3 ��p�V����T���3��.^"�~n2���m[�Ϥ ��� A&���H��'�~�V>�m"�t� �?�� �P۷�� �,�.��;�� ��UbPV����c���H�^1��o��'z|���8���鵓��$ӯ%���$�����>�<+#l�I�� �n=T��9���"��|l�o( Ql;�O+�E��C���nɵ�K և�ƪ٭�Af� &@� �� ���8F� ��� ��g�v�cМu�͘)a%qKo�0��w��K!�Uˇ��6�� ;:�����]d�g*��e4Bd�h����;� Nf���{$�Yvb�~�����K�ۂ=/ ��g7�-���q�^�̼3�Q}ɶ�1���}yZ�� �OA ���]*���^剂�` �"�^Gh��� �SdV�D^1N~ӛc��`�n������A ���qM�~rp� s����� *� u�? w�SAOsv��Ү�m�c�����\k��VΥ�M� � _ �]Y7�|X�) ���?�-�;֓��Xm�ׯ@��kH��n����%�<Jg6j�C:$x�<��3�A���ԝdc9 01��d9X�C���uq �y���r��D4ˈfU�� dAb� �#��\�c��/�)���OlnE-�i̙�+F/y��F?qe������ۚ>H�KyÇ�f����̫4��� �� Pnޭ��6Ԉk�܌cD �}�@J7� R�N�� ��j���LIl� m���"{��/��;��K۹R��,�P@�9ф9�G)Ǹ� &���I�n3geȽ}V����E3}͙؇��s8v�k��DQ�]�� ����g��lҌ�Ю�/_� |%E ��F��I�uJ ��~��Ŗw!���<�y!�~9���aS Y��pb�o�/M�ෟO�3Y��+S)垵��|>�p}`�m� B|��%��ET��So9��[����5~P��b���ڪ� ��W�R Kσ.���Y#���:5�����| <ɳ�~P O���d�2�2N�� U�h^-R�S ���^� <�� ���Ր�܊&A �Q<�W% � )�_�%e\{�����9�a� ��Z$�por�̎� g��a>��&?އ:(}� T��q|� ��l���.],��]DD���5 r����xj�/M=e��9�tM��l�|�~Rp�%a%}:lj<�n�5�����y����"}/�d��;l��w��p�|R�K�4�nc�6�KPI�80.?Gz�|۾*ӌ����ʻ"�0�ԑ B���Џ���7��?_�F���텳05~ۓ��@ ��Wx��v** �h#�@�cL��у(hd|����1V��5R�nl~�$�|��'2���oCվ67[�C.� �f#�{�l�*a�5 _��_ �ޕ�-¶�n1+�7����bU�:��¢����H$�# �o[���M��?�n�G �5^D�\=,�bi7C �� ��$9"%t9D�b[^�� ԝ �,m��?sT2�C��]w/k��QX�+�y��& ^��#g �OƗ��N�7f) �#�8�U�>�� �2<�/ڤ�aO ��XqI0^�ְ�@ !J� +w�&{��C�}�ְ��Ü�1���vN���O]}π�ߚY�R^j�Ά�:�S�>'��׳����:<s�X�������� �i˳b(�G�|�Af�S�]:A��%n��/��vN��Wl�O��Diҁ�]�s�gKW�͢�%��z���� o=��� �"��Q�s]��8�<� x�'dۢ�_��c��p�x�>�+Z��|� ]�����V�z�� ����գ��]H ��)8˥��p��z,�IV f$�F�į�����u �bt[n�^�H!�C�k��ո��G9��P]�{џ\;�9̓ K��p���/ Ԗ~;���r qr� g�� h���j�;q� �{Z $��sg���H�c%�xP�j<'⇃�v�3w�}�d�E�pq���~�W4_F+2,<��4�K�j��/�K�KQj�ؾn�V���拃ǡZ(��� ��J���R ��P?!�1o&�'�A# �)ì ���8��变�^sc>��A˻ b@���:���=2W��ҘU� ��ً<��8 R^BN&��2 .E�Էg�Sn~r��q�OD8��xn�����_4m@��� j��UX7�+1r���텳�q�l���K^�pLӈp����"S��{H+e�hs ��)g���u�b��1zK��r��=� ��:�ho���*��Fg��ChBl��y�o��{"Ds��4 �,*s�i�M��,(е�w�����m� ��-y� cM�J��ޛ 3;�cJ3�mZ��SAt�Or�� �O����8ڳ>I"���Pw��06h< ��C��d:�A;��Cg2��5[���GwtR� �Q���Q����]B*��!�<ژݡ���E'�Y�<M �8���� �ZC�e�z(��*8�w����6 ��4��\Q��<N*��ׅJ�\��x��yN]l[ �kӏ0� $�)д��� ��� A�� hc/��� ڻz8��%�x��N]�@��&֭�J��'��2�?��fu�T;���/R�V-��7�cn1+� ���;R�#x���[q�� ��^�w]�9 4�J7�X� L�R%��8JSF1��+�T������ZNC֎���aX�N\vYt�y�Y뚝S���ɔ<h�����;?TS!%A�SAY�s�;�o �x�QQ �%�vQ �^�j[�� o�W٩��U���I��b�)����h�ӮF!a��ʩ�Va� �����T%G�n�;�m2��!gz��b�_{ہo��Ж��I9��۶�d�Zp#p[B,9(Ї N<t���~h|��e���HѫM� �������^z%c�x���v�2��Y)�ѹ*(j��ĺ�%�yq�v!̢�I,W��婡�c#�@S�K��1ߺ5ژ� x��������K��� E�Fp��T[_�O�Sr�SW�=�J�B ui j>� ��AsC�ۚg���?��L�]��;�!���.���ޗ��u��&�4��Ѿ�u� ��g�o�+V�O�T�w�.�2���M� �oT"5i����c�CmQB!�v�HE���O�hj�ZR��!"~;�M���>�,���`{�+���䭮�t����B#Z.0<G M#P��eF��u�8 ��ԎiV���g�7�(���y��n�����c ���ր�� �n�vlI�;��w� �J������ �Þ���!��.o2��X1���K �ym�}�ew"�J 1�ί�u��&(���#�����? �_��{�� vc ����B�tFɊ� ��`'�v'w�E� ��_;d��5Yxؑ/�������X;��� *�\�L��35��cWR���e�ޢ$ s�ګ}o;6�N?n�n?j�Zh�[|Xdu�2c��F�W�`bP����HDmu��D�1�~g>Q�(o������u�� �_��d���GD���٤\A]{=�++��XpC ���8 Io��@�8�N�} 9�2�j'^"�&���ȝ���Ͻ��*+.l���#ץ��n��%o��+N��l�Y���㼨�ϙ"b}��..�UX�Bھ���#7v) �JSX ���j���`�5;xv�����(���~����� ��:��\�@�w� s52��Yvicҵ<�aˋ��$=��ɾv�v��l-~�l��1�a^�' ����\mV�Ձ��Vk�S:oހ��ůp�|[dV�7 w1Or�� ��H�p���UI'r p��h��ˋR^�������݇��W? �sO�5}J��� ���>$e!�Ԇ� �žK� a ��X�3�+�25!�#&;�l���-�,�J6�=����}q{�q �5¢t�j��{#f <h=τ�p?C��+%�7�It�@8H�h8�HV3�:k¢��v��4���; ��g� �d����Yn��1�<J�{��l�QΓ�wt�ay�@@K0��f����}3f��|���?`�I `=�%b��)M� �&f:% �rk�nM͠s�4��V�i_�en7b}��E��X�M1G�|~U���V�C���P X5VT��r��\���J7�s>��-e����l��k?N��p���Nxg*b{s;�Ϗ�Q�9�q��% j {/h%�VT��^ ���� ($��}�����\�q ��ֱ�e �����-v� b��.�^ v=�f0ɾ�=��w�٥�[ `��6 �Ũ��M;��'a�k u$�m� %y�79����G ���",��U&�YNѓ9�JUQ2.�Fc���u4lH���Ο��L!$�m^꠩�9��r�x�ΨD��� ����#;H�P� ��]u� 3��øY�'�L/}k���|���s痀�)� :�=����fN=[��,I���̀#�nq�sy �_�%��!�����"eW#j�<Oֻd��.@V� iu �0�|$��]�� �fڗ?혱��������ɊG�F��w*��&� �Q4t�Z5K�{�a��[l���=��� �K4h�������k��@ Ά�-`��1����Oq�c�{RX8��� N�(s�DZ��*����5��z>ǂ����{ii�R֚22� i���mY��D��i��hP�B.��c�� PW�O�"� ������p9,��k�}�h^{��� z�?��ޙ �.���|Mϧ9G���.e�x(�0r��6��")��IcJl������[W n�����]��Y �Axj&�u�D�3r�b&��K���Z�f^����:����H~�,"� y�+}�ڈrvy��W��?e��db�wFG���j���d��,�\bД-Y�]%NC� y��n���Ѯ$^��_:'��Vm�\g�P�˭qړ�����k� 5[f ���% ' stqߪ�B��8�:�C�'s(�D��(���ۻ��L�i4��U'lK7fP� #|i��B��k$��a$Z�<}sp�^�e^$�Y��̑6��Ѭ�QiH8� 1�jI�/4B�/��74���b�Ϊ��A5YD���/!�8i����p�-o^uCuJ�f��c�2����?ʟR* Y�j�t�."��������Z-N�N�grLU���g�\�m�v"~Wo�"��� +�����q,��gT�E�iR�1�:Z"�i�&�Dz̜�M� \�>�����t���s�Tƶ�e5� "�u�P� ����� �1�g�p<v� ��6.�R�?x��1� 3ѻ)��m�ͻ܋ G,�lG���M@=j�a\�s��, le7E;i|�� v.�Ld� ��jZ�s���Hvz��xN� � �Tl�~���0 �UzgG�`;���*W^[�XH�{�Nb��rW���n�h��Fdf YF�!�@� �M�{��ٱK)g��ʻI�r:��'1�>�%�p��u Ĺ��FE��d��C�]~ɡ�\n���n!Mv�*�B|tdS!�J����� 0��W�Q7� �n �z�M>g[@{�<U��" M��|���AI����W��^m����"� U�P� �~r�fQB�����'�!:mW�Ѳ�/6̽�!?��/�N�q�B��Z�o�xeY��� �@)�e$ ���.W system_tests_sync/.gitignore 0000644 00000000020 15025175721 0012333 0 ustar 00 data secrets.tar system_tests_sync/test_grpc.py 0000644 00000006671 15025175721 0012731 0 ustar 00 # Copyright 2016 Google LLC # # 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. import google.auth import google.auth.credentials import google.auth.jwt import google.auth.transport.grpc from google.oauth2 import service_account from google.cloud import pubsub_v1 def test_grpc_request_with_regular_credentials(http_request): credentials, project_id = google.auth.default() credentials = google.auth.credentials.with_scopes_if_required( credentials, scopes=["https://www.googleapis.com/auth/pubsub"] ) # Create a pub/sub client. client = pubsub_v1.PublisherClient(credentials=credentials) # list the topics and drain the iterator to test that an authorized API # call works. list_topics_iter = client.list_topics(project="projects/{}".format(project_id)) list(list_topics_iter) def test_grpc_request_with_regular_credentials_and_self_signed_jwt(http_request): credentials, project_id = google.auth.default() # At the time this test is being written, there are no GAPIC libraries # that will trigger the self-signed JWT flow. Manually create the self-signed # jwt on the service account credential to check that the request # succeeds. credentials = credentials.with_scopes( scopes=[], default_scopes=["https://www.googleapis.com/auth/pubsub"] ) credentials._create_self_signed_jwt(audience="https://pubsub.googleapis.com/") # Create a pub/sub client. client = pubsub_v1.PublisherClient(credentials=credentials) # list the topics and drain the iterator to test that an authorized API # call works. list_topics_iter = client.list_topics(project="projects/{}".format(project_id)) list(list_topics_iter) # Check that self-signed JWT was created and is being used assert credentials._jwt_credentials is not None assert credentials._jwt_credentials.token == credentials.token def test_grpc_request_with_jwt_credentials(): credentials, project_id = google.auth.default() audience = "https://pubsub.googleapis.com/google.pubsub.v1.Publisher" credentials = google.auth.jwt.Credentials.from_signing_credentials( credentials, audience=audience ) # Create a pub/sub client. client = pubsub_v1.PublisherClient(credentials=credentials) # list the topics and drain the iterator to test that an authorized API # call works. list_topics_iter = client.list_topics(project="projects/{}".format(project_id)) list(list_topics_iter) def test_grpc_request_with_on_demand_jwt_credentials(): credentials, project_id = google.auth.default() credentials = google.auth.jwt.OnDemandCredentials.from_signing_credentials( credentials ) # Create a pub/sub client. client = pubsub_v1.PublisherClient(credentials=credentials) # list the topics and drain the iterator to test that an authorized API # call works. list_topics_iter = client.list_topics(project="projects/{}".format(project_id)) list(list_topics_iter) system_tests_sync/test_requests.py 0000644 00000003104 15025175721 0013635 0 ustar 00 # Copyright 2021 Google LLC # # 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. import google.auth import google.auth.credentials import google.auth.transport.requests from google.oauth2 import service_account def test_authorized_session_with_service_account_and_self_signed_jwt(): credentials, project_id = google.auth.default() credentials = credentials.with_scopes( scopes=[], default_scopes=["https://www.googleapis.com/auth/pubsub"], ) session = google.auth.transport.requests.AuthorizedSession( credentials=credentials, default_host="pubsub.googleapis.com" ) # List Pub/Sub Topics through the REST API # https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics/list url = "https://pubsub.googleapis.com/v1/projects/{}/topics".format(project_id) with session: response = session.get(url) response.raise_for_status() # Check that self-signed JWT was created and is being used assert credentials._jwt_credentials is not None assert credentials._jwt_credentials.token.decode() == credentials.token system_tests_sync/test_default.py 0000644 00000001565 15025175721 0013417 0 ustar 00 # Copyright 2016 Google LLC # # 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. import os import google.auth EXPECT_PROJECT_ID = os.environ.get("EXPECT_PROJECT_ID") def test_application_default_credentials(verify_refresh): credentials, project_id = google.auth.default() if EXPECT_PROJECT_ID is not None: assert project_id is not None verify_refresh(credentials) system_tests_sync/test_urllib3.py 0000644 00000003101 15025175721 0013333 0 ustar 00 # Copyright 2021 Google LLC # # 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. import google.auth import google.auth.credentials import google.auth.transport.requests from google.oauth2 import service_account def test_authorized_session_with_service_account_and_self_signed_jwt(): credentials, project_id = google.auth.default() credentials = credentials.with_scopes( scopes=[], default_scopes=["https://www.googleapis.com/auth/pubsub"], ) http = google.auth.transport.urllib3.AuthorizedHttp( credentials=credentials, default_host="pubsub.googleapis.com" ) # List Pub/Sub Topics through the REST API # https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.topics/list response = http.urlopen( method="GET", url="https://pubsub.googleapis.com/v1/projects/{}/topics".format(project_id) ) assert response.status == 200 # Check that self-signed JWT was created and is being used assert credentials._jwt_credentials is not None assert credentials._jwt_credentials.token.decode() == credentials.token system_tests_sync/test_compute_engine.py 0000644 00000004514 15025175721 0014771 0 ustar 00 # Copyright 2016 Google LLC # # 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. from datetime import datetime import pytest import google.auth from google.auth import compute_engine from google.auth import _helpers from google.auth import exceptions from google.auth import jwt from google.auth.compute_engine import _metadata import google.oauth2.id_token AUDIENCE = "https://pubsub.googleapis.com" @pytest.fixture(autouse=True) def check_gce_environment(http_request): try: _metadata.get_service_account_info(http_request) except exceptions.TransportError: pytest.skip("Compute Engine metadata service is not available.") def test_refresh(http_request, token_info): credentials = compute_engine.Credentials() credentials.refresh(http_request) assert credentials.token is not None assert credentials.service_account_email is not None info = token_info(credentials.token) info_scopes = _helpers.string_to_scopes(info["scope"]) assert set(info_scopes) == set(credentials.scopes) def test_default(verify_refresh): credentials, project_id = google.auth.default() assert project_id is not None assert isinstance(credentials, compute_engine.Credentials) verify_refresh(credentials) def test_id_token_from_metadata(http_request): credentials = compute_engine.IDTokenCredentials( http_request, AUDIENCE, use_metadata_identity_endpoint=True ) credentials.refresh(http_request) _, payload, _, _ = jwt._unverified_decode(credentials.token) assert credentials.valid assert payload["aud"] == AUDIENCE assert datetime.fromtimestamp(payload["exp"]) == credentials.expiry def test_fetch_id_token(http_request): token = google.oauth2.id_token.fetch_id_token(http_request, AUDIENCE) _, payload, _, _ = jwt._unverified_decode(token) assert payload["aud"] == AUDIENCE system_tests_sync/test_id_token.py 0000644 00000001606 15025175721 0013563 0 ustar 00 # Copyright 2020 Google LLC # # 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. import pytest from google.auth import jwt import google.oauth2.id_token def test_fetch_id_token(http_request): audience = "https://pubsub.googleapis.com" token = google.oauth2.id_token.fetch_id_token(http_request, audience) _, payload, _, _ = jwt._unverified_decode(token) assert payload["aud"] == audience system_tests_sync/test_service_account.py 0000644 00000003650 15025175721 0015144 0 ustar 00 # Copyright 2016 Google LLC # # 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. import pytest from google.auth import _helpers from google.auth import exceptions from google.auth import iam from google.oauth2 import service_account @pytest.fixture def credentials(service_account_file): yield service_account.Credentials.from_service_account_file(service_account_file) def test_refresh_no_scopes(http_request, credentials): with pytest.raises(exceptions.RefreshError): credentials.refresh(http_request) def test_refresh_success(http_request, credentials, token_info): credentials = credentials.with_scopes(["email", "profile"]) credentials.refresh(http_request) assert credentials.token info = token_info(credentials.token) assert info["email"] == credentials.service_account_email info_scopes = _helpers.string_to_scopes(info["scope"]) assert set(info_scopes) == set( [ "https://www.googleapis.com/auth/userinfo.email", "https://www.googleapis.com/auth/userinfo.profile", ] ) def test_iam_signer(http_request, credentials): credentials = credentials.with_scopes( ["https://www.googleapis.com/auth/iam"] ) # Verify iamcredentials signer. signer = iam.Signer( http_request, credentials, credentials.service_account_email ) signed_blob = signer.sign("message") assert isinstance(signed_blob, bytes) system_tests_sync/test_impersonated_credentials.py 0000644 00000006567 15025175721 0017051 0 ustar 00 # Copyright 2020 Google LLC # # 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. import json import pytest import google.oauth2.credentials from google.oauth2 import service_account import google.auth.impersonated_credentials from google.auth import _helpers GOOGLE_OAUTH2_TOKEN_ENDPOINT = "https://oauth2.googleapis.com/token" @pytest.fixture def service_account_credentials(service_account_file): yield service_account.Credentials.from_service_account_file(service_account_file) @pytest.fixture def impersonated_service_account_credentials(impersonated_service_account_file): yield service_account.Credentials.from_service_account_file( impersonated_service_account_file ) def test_refresh_with_user_credentials_as_source( authorized_user_file, impersonated_service_account_credentials, http_request, token_info, ): with open(authorized_user_file, "r") as fh: info = json.load(fh) source_credentials = google.oauth2.credentials.Credentials( None, refresh_token=info["refresh_token"], token_uri=GOOGLE_OAUTH2_TOKEN_ENDPOINT, client_id=info["client_id"], client_secret=info["client_secret"], # The source credential needs this scope for the generateAccessToken request # The user must also have `Service Account Token Creator` on the project # that owns the impersonated service account. # See https://cloud.google.com/iam/docs/creating-short-lived-service-account-credentials scopes=["https://www.googleapis.com/auth/cloud-platform"], ) source_credentials.refresh(http_request) target_scopes = [ "https://www.googleapis.com/auth/devstorage.read_only", "https://www.googleapis.com/auth/analytics", ] target_credentials = google.auth.impersonated_credentials.Credentials( source_credentials=source_credentials, target_principal=impersonated_service_account_credentials.service_account_email, target_scopes=target_scopes, lifetime=100, ) target_credentials.refresh(http_request) assert target_credentials.token def test_refresh_with_service_account_credentials_as_source( http_request, service_account_credentials, impersonated_service_account_credentials, token_info, ): source_credentials = service_account_credentials.with_scopes(["email"]) source_credentials.refresh(http_request) assert source_credentials.token target_scopes = [ "https://www.googleapis.com/auth/devstorage.read_only", "https://www.googleapis.com/auth/analytics", ] target_credentials = google.auth.impersonated_credentials.Credentials( source_credentials=source_credentials, target_principal=impersonated_service_account_credentials.service_account_email, target_scopes=target_scopes, ) target_credentials.refresh(http_request) assert target_credentials.token system_tests_sync/test_external_accounts.py 0000644 00000040162 15025175721 0015510 0 ustar 00 # Copyright 2021 Google LLC # # 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. # Prerequisites: # Make sure to run the setup in scripts/setup_external_accounts.sh # and copy the logged constant strings (_AUDIENCE_OIDC, _AUDIENCE_AWS) # into this file before running this test suite. # Once that is done, this test can be run indefinitely. # # The only requirement for this test suite to run is to set the environment # variable GOOGLE_APPLICATION_CREDENTIALS to point to the expected service # account keys whose email is referred to in the setup script. # # This script follows the following logic. # OIDC provider (file-sourced and url-sourced credentials): # Use the service account keys to generate a Google ID token using the # iamcredentials generateIdToken API, using the default STS audience. # This will use the service account client ID as the sub field of the token. # This OIDC token will be used as the external subject token to be exchanged # for a Google access token via GCP STS endpoint and then to impersonate the # original service account key. import datetime import json import os import socket from tempfile import NamedTemporaryFile import threading import time import sys import google.auth from google.auth import _helpers from googleapiclient import discovery from http.server import BaseHTTPRequestHandler from http.server import HTTPServer from google.oauth2 import service_account import pytest from mock import patch # Populate values from the output of scripts/setup_external_accounts.sh. _AUDIENCE_OIDC = "//iam.googleapis.com/projects/79992041559/locations/global/workloadIdentityPools/pool-73wslmxn/providers/oidc-73wslmxn" _AUDIENCE_AWS = "//iam.googleapis.com/projects/79992041559/locations/global/workloadIdentityPools/pool-73wslmxn/providers/aws-73wslmxn" _ROLE_AWS = "arn:aws:iam::077071391996:role/ci-python-test" def dns_access_direct(request, project_id): # First, get the default credentials. credentials, _ = google.auth.default( scopes=["https://www.googleapis.com/auth/cloud-platform.read-only"], request=request, ) # Apply the default credentials to the headers to make the request. headers = {} credentials.apply(headers) response = request( url="https://dns.googleapis.com/dns/v1/projects/{}".format(project_id), headers=headers, ) if response.status == 200: return response.data def dns_access_client_library(_, project_id): service = discovery.build("dns", "v1") request = service.projects().get(project=project_id) return request.execute() @pytest.fixture(params=[dns_access_direct, dns_access_client_library]) def dns_access(request, http_request, service_account_info): # Fill in the fixtures on the functions, # so that we don't have to fill in the parameters manually. def wrapper(): return request.param(http_request, service_account_info["project_id"]) yield wrapper @pytest.fixture def oidc_credentials(service_account_file, http_request): result = service_account.IDTokenCredentials.from_service_account_file( service_account_file, target_audience=_AUDIENCE_OIDC ) result.refresh(http_request) yield result @pytest.fixture def service_account_info(service_account_file): with open(service_account_file) as f: yield json.load(f) @pytest.fixture def aws_oidc_credentials( service_account_file, service_account_info, authenticated_request ): credentials = service_account.Credentials.from_service_account_file( service_account_file, scopes=["https://www.googleapis.com/auth/cloud-platform"] ) result = authenticated_request(credentials)( url="https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/{}:generateIdToken".format( service_account_info["client_email"] ), method="POST", body=json.dumps( {"audience": service_account_info["client_id"], "includeEmail": True} ), ) assert result.status == 200 yield json.loads(result.data)["token"] # Our external accounts tests involve setting up some preconditions, setting a # credential file, and then making sure that our client libraries can work with # the set credentials. def get_project_dns(dns_access, credential_data): with NamedTemporaryFile() as credfile: credfile.write(json.dumps(credential_data).encode("utf-8")) credfile.flush() with patch.dict(os.environ, {"GOOGLE_APPLICATION_CREDENTIALS": credfile.name}): # If our setup and credential file are correct, # discovery.build should be able to establish these as the default credentials. return dns_access() def get_xml_value_by_tagname(data, tagname): startIndex = data.index("<{}>".format(tagname)) if startIndex >= 0: endIndex = data.index("</{}>".format(tagname), startIndex) if endIndex > startIndex: return data[startIndex + len(tagname) + 2 : endIndex] # This test makes sure that setting an accesible credential file # works to allow access to Google resources. def test_file_based_external_account(oidc_credentials, dns_access): with NamedTemporaryFile() as tmpfile: tmpfile.write(oidc_credentials.token.encode("utf-8")) tmpfile.flush() assert get_project_dns( dns_access, { "type": "external_account", "audience": _AUDIENCE_OIDC, "subject_token_type": "urn:ietf:params:oauth:token-type:jwt", "token_url": "https://sts.googleapis.com/v1/token", "service_account_impersonation_url": "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/{}:generateAccessToken".format( oidc_credentials.service_account_email ), "credential_source": { "file": tmpfile.name, }, }, ) # This test makes sure that setting a token lifetime works # for service account impersonation. def test_file_based_external_account_with_configure_token_lifetime( oidc_credentials, dns_access ): with NamedTemporaryFile() as tmpfile: tmpfile.write(oidc_credentials.token.encode("utf-8")) tmpfile.flush() assert get_project_dns( dns_access, { "type": "external_account", "audience": _AUDIENCE_OIDC, "subject_token_type": "urn:ietf:params:oauth:token-type:jwt", "token_url": "https://sts.googleapis.com/v1/token", "service_account_impersonation_url": "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/{}:generateAccessToken".format( oidc_credentials.service_account_email ), "service_account_impersonation": { "token_lifetime_seconds": 2800, }, "credential_source": { "file": tmpfile.name, }, }, ) def test_configurable_token_lifespan(oidc_credentials, http_request): TOKEN_LIFETIME_SECONDS = 2800 BUFFER_SECONDS = 5 def check_impersonation_expiration(): # First, get the default credentials. credentials, _ = google.auth.default( scopes=["https://www.googleapis.com/auth/cloud-platform.read-only"], request=http_request, ) utcmax = _helpers.utcnow() + datetime.timedelta(seconds=TOKEN_LIFETIME_SECONDS) utcmin = utcmax - datetime.timedelta(seconds=BUFFER_SECONDS) assert utcmin < credentials._impersonated_credentials.expiry <= utcmax return True with NamedTemporaryFile() as tmpfile: tmpfile.write(oidc_credentials.token.encode("utf-8")) tmpfile.flush() assert get_project_dns( check_impersonation_expiration, { "type": "external_account", "audience": _AUDIENCE_OIDC, "subject_token_type": "urn:ietf:params:oauth:token-type:jwt", "token_url": "https://sts.googleapis.com/v1/token", "service_account_impersonation_url": "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/{}:generateAccessToken".format( oidc_credentials.service_account_email ), "service_account_impersonation": { "token_lifetime_seconds": TOKEN_LIFETIME_SECONDS, }, "credential_source": { "file": tmpfile.name, }, }, ) # This test makes sure that setting up an http server to provide credentials # works to allow access to Google resources. def test_url_based_external_account(dns_access, oidc_credentials, service_account_info): class TestResponseHandler(BaseHTTPRequestHandler): def do_GET(self): if self.headers["my-header"] != "expected-value": self.send_response(400) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write( json.dumps({"error": "missing header"}).encode("utf-8") ) elif self.path != "/token": self.send_response(400) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write( json.dumps({"error": "incorrect token path"}).encode("utf-8") ) else: self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write( json.dumps({"access_token": oidc_credentials.token}).encode("utf-8") ) class TestHTTPServer(HTTPServer, object): def __init__(self): self.port = self._find_open_port() super(TestHTTPServer, self).__init__(("", self.port), TestResponseHandler) @staticmethod def _find_open_port(): s = socket.socket() s.bind(("", 0)) return s.getsockname()[1] # This makes sure that the server gets shut down when this variable leaves its "with" block # The python3 HttpServer has __enter__ and __exit__ methods, but python2 does not. # By redefining the __enter__ and __exit__ methods, we ensure that python2 and python3 act similarly def __exit__(self, *args): self.shutdown() def __enter__(self): return self with TestHTTPServer() as server: threading.Thread(target=server.serve_forever).start() assert get_project_dns( dns_access, { "type": "external_account", "audience": _AUDIENCE_OIDC, "subject_token_type": "urn:ietf:params:oauth:token-type:jwt", "token_url": "https://sts.googleapis.com/v1/token", "service_account_impersonation_url": "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/{}:generateAccessToken".format( oidc_credentials.service_account_email ), "credential_source": { "url": "http://localhost:{}/token".format(server.port), "headers": {"my-header": "expected-value"}, "format": { "type": "json", "subject_token_field_name": "access_token", }, }, }, ) # AWS provider tests for AWS credentials # The test suite will also run tests for AWS credentials. This works as # follows. (Note prequisite setup is needed. This is documented in # setup_external_accounts.sh). # - iamcredentials:generateIdToken is used to generate a Google ID token using # the service account access token. The service account client_id is used as # audience. # - AWS STS AssumeRoleWithWebIdentity API is used to exchange this token for # temporary AWS security credentials for a specified AWS ARN role. # - AWS_REGION, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and AWS_SESSION_TOKEN # environment variables are set using these credentials before the test is # run simulating an AWS VM. # - The test can now be run. def test_aws_based_external_account( aws_oidc_credentials, service_account_info, dns_access, http_request ): # temporarily disable this test as we investigate why it fails # https://github.com/googleapis/google-auth-library-python/issues/1279 return response = http_request( url=( "https://sts.amazonaws.com/" "?Action=AssumeRoleWithWebIdentity" "&Version=2011-06-15" "&DurationSeconds=3600" "&RoleSessionName=python-test" "&RoleArn={}" "&WebIdentityToken={}" ).format(_ROLE_AWS, aws_oidc_credentials) ) assert response.status == 200 # The returned data is in XML, but loading an XML parser would be overkill. # Searching the return text manually for the start and finish tag. data = response.data.decode("utf-8") with patch.dict( os.environ, { "AWS_REGION": "us-east-2", "AWS_ACCESS_KEY_ID": get_xml_value_by_tagname(data, "AccessKeyId"), "AWS_SECRET_ACCESS_KEY": get_xml_value_by_tagname(data, "SecretAccessKey"), "AWS_SESSION_TOKEN": get_xml_value_by_tagname(data, "SessionToken"), }, ): assert get_project_dns( dns_access, { "type": "external_account", "audience": _AUDIENCE_AWS, "subject_token_type": "urn:ietf:params:aws:token-type:aws4_request", "token_url": "https://sts.googleapis.com/v1/token", "service_account_impersonation_url": "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/{}:generateAccessToken".format( service_account_info["client_email"] ), "credential_source": { "environment_id": "aws1", "regional_cred_verification_url": "https://sts.{region}.amazonaws.com?Action=GetCallerIdentity&Version=2011-06-15", }, }, ) # This test makes sure that setting up an executable to provide credentials # works to allow access to Google resources. def test_pluggable_external_account(oidc_credentials, service_account_info, dns_access): now = datetime.datetime.now() unix_seconds = time.mktime(now.timetuple()) expiration_time = (unix_seconds + 1 * 60 * 60) * 1000 credential = { "success": True, "version": 1, "expiration_time": expiration_time, "token_type": "urn:ietf:params:oauth:token-type:jwt", "id_token": oidc_credentials.token, } tmpfile = NamedTemporaryFile(delete=True) with open(tmpfile.name, "w") as f: f.write("#!/bin/bash\n") f.write('echo "{}"\n'.format(json.dumps(credential).replace('"', '\\"'))) tmpfile.file.close() os.chmod(tmpfile.name, 0o777) assert get_project_dns( dns_access, { "type": "external_account", "audience": _AUDIENCE_OIDC, "subject_token_type": "urn:ietf:params:oauth:token-type:jwt", "token_url": "https://sts.googleapis.com/v1/token", "service_account_impersonation_url": "https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/{}:generateAccessToken".format( oidc_credentials.service_account_email ), "credential_source": { "executable": { "command": tmpfile.name, } }, }, ) system_tests_sync/conftest.py 0000644 00000010051 15025175721 0012547 0 ustar 00 # Copyright 2016 Google LLC # # 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. import json import os from google.auth import _helpers import google.auth.transport.requests import google.auth.transport.urllib3 import pytest import requests import urllib3 HERE = os.path.dirname(__file__) DATA_DIR = os.path.join(HERE, "../data") IMPERSONATED_SERVICE_ACCOUNT_FILE = os.path.join( DATA_DIR, "impersonated_service_account.json" ) SERVICE_ACCOUNT_FILE = os.path.join(DATA_DIR, "service_account.json") URLLIB3_HTTP = urllib3.PoolManager(retries=False) REQUESTS_SESSION = requests.Session() REQUESTS_SESSION.verify = False TOKEN_INFO_URL = "https://www.googleapis.com/oauth2/v3/tokeninfo" @pytest.fixture def service_account_file(): """The full path to a valid service account key file.""" yield SERVICE_ACCOUNT_FILE @pytest.fixture def impersonated_service_account_file(): """The full path to a valid service account key file.""" yield IMPERSONATED_SERVICE_ACCOUNT_FILE @pytest.fixture def authorized_user_file(): """The full path to a valid authorized user file.""" yield AUTHORIZED_USER_FILE @pytest.fixture(params=["urllib3", "requests"]) def request_type(request): yield request.param @pytest.fixture def http_request(request_type): """A transport.request object.""" if request_type == "urllib3": yield google.auth.transport.urllib3.Request(URLLIB3_HTTP) elif request_type == "requests": yield google.auth.transport.requests.Request(REQUESTS_SESSION) @pytest.fixture def authenticated_request(request_type): """A transport.request object that takes credentials""" if request_type == "urllib3": def wrapper(credentials): return google.auth.transport.urllib3.AuthorizedHttp( credentials, http=URLLIB3_HTTP ).request yield wrapper elif request_type == "requests": def wrapper(credentials): session = google.auth.transport.requests.AuthorizedSession(credentials) session.verify = False return google.auth.transport.requests.Request(session) yield wrapper @pytest.fixture def token_info(http_request): """Returns a function that obtains OAuth2 token info.""" def _token_info(access_token=None, id_token=None): query_params = {} if access_token is not None: query_params["access_token"] = access_token elif id_token is not None: query_params["id_token"] = id_token else: raise ValueError("No token specified.") url = _helpers.update_query(TOKEN_INFO_URL, query_params) response = http_request(url=url, method="GET") return json.loads(response.data.decode("utf-8")) yield _token_info @pytest.fixture def verify_refresh(http_request): """Returns a function that verifies that credentials can be refreshed.""" def _verify_refresh(credentials): if credentials.requires_scopes: credentials = credentials.with_scopes(["email", "profile"]) credentials.refresh(http_request) assert credentials.token assert credentials.valid yield _verify_refresh def verify_environment(): """Checks to make sure that requisite data files are available.""" if not os.path.isdir(DATA_DIR): raise EnvironmentError( "In order to run system tests, test data must exist in " "system_tests/data. See CONTRIBUTING.rst for details." ) def pytest_configure(config): """Pytest hook that runs before Pytest collects any tests.""" verify_environment() system_tests_sync/test_downscoping.py 0000644 00000015047 15025175721 0014325 0 ustar 00 # Copyright 2021 Google LLC # # 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. import re import uuid import google.auth from google.auth import downscoped from google.auth.transport import requests from google.cloud import exceptions from google.cloud import storage from google.oauth2 import credentials import pytest # The object prefix used to test access to files beginning with this prefix. _OBJECT_PREFIX = "customer-a" # The object name of the object inaccessible by the downscoped token. _ACCESSIBLE_OBJECT_NAME = "{0}-data.txt".format(_OBJECT_PREFIX) # The content of the object accessible by the downscoped token. _ACCESSIBLE_CONTENT = "hello world" # The content of the object inaccessible by the downscoped token. _INACCESSIBLE_CONTENT = "secret content" # The object name of the object inaccessible by the downscoped token. _INACCESSIBLE_OBJECT_NAME = "other-customer-data.txt" @pytest.fixture(scope="module") def temp_bucket(): """Yields a bucket that is deleted after the test completes.""" bucket = None while bucket is None or bucket.exists(): bucket_name = "auth-python-downscope-test-{}".format(uuid.uuid4()) bucket = storage.Client().bucket(bucket_name) bucket = storage.Client().create_bucket(bucket.name) yield bucket bucket.delete(force=True) @pytest.fixture(scope="module") def temp_blobs(temp_bucket): """Yields two blobs that are deleted after the test completes.""" bucket = temp_bucket # Downscoped tokens will have readonly access to this blob. accessible_blob = bucket.blob(_ACCESSIBLE_OBJECT_NAME) accessible_blob.upload_from_string(_ACCESSIBLE_CONTENT) # Downscoped tokens will have no access to this blob. inaccessible_blob = bucket.blob(_INACCESSIBLE_OBJECT_NAME) inaccessible_blob.upload_from_string(_INACCESSIBLE_CONTENT) yield (accessible_blob, inaccessible_blob) bucket.delete_blobs([accessible_blob, inaccessible_blob]) def get_token_from_broker(bucket_name, object_prefix): """Simulates token broker generating downscoped tokens for specified bucket. Args: bucket_name (str): The name of the Cloud Storage bucket. object_prefix (str): The prefix string of the object name. This is used to ensure access is restricted to only objects starting with this prefix string. Returns: Tuple[str, datetime.datetime]: The downscoped access token and its expiry date. """ # Initialize the Credential Access Boundary rules. available_resource = "//storage.googleapis.com/projects/_/buckets/{0}".format(bucket_name) # Downscoped credentials will have readonly access to the resource. available_permissions = ["inRole:roles/storage.objectViewer"] # Only objects starting with the specified prefix string in the object name # will be allowed read access. availability_expression = ( "resource.name.startsWith('projects/_/buckets/{0}/objects/{1}')".format(bucket_name, object_prefix) ) availability_condition = downscoped.AvailabilityCondition(availability_expression) # Define the single access boundary rule using the above properties. rule = downscoped.AccessBoundaryRule( available_resource=available_resource, available_permissions=available_permissions, availability_condition=availability_condition, ) # Define the Credential Access Boundary with all the relevant rules. credential_access_boundary = downscoped.CredentialAccessBoundary(rules=[rule]) # Retrieve the source credentials via ADC. source_credentials, _ = google.auth.default() if source_credentials.requires_scopes: source_credentials = source_credentials.with_scopes( ["https://www.googleapis.com/auth/cloud-platform"] ) # Create the downscoped credentials. downscoped_credentials = downscoped.Credentials( source_credentials=source_credentials, credential_access_boundary=credential_access_boundary, ) # Refresh the tokens. downscoped_credentials.refresh(requests.Request()) # These values will need to be passed to the token consumer. access_token = downscoped_credentials.token expiry = downscoped_credentials.expiry return (access_token, expiry) def test_downscoping(temp_blobs): """Tests token consumer access to cloud storage using downscoped tokens. Args: temp_blobs (Tuple[google.cloud.storage.blob.Blob, ...]): The temporarily created test cloud storage blobs (one readonly accessible, the other not). """ accessible_blob, inaccessible_blob = temp_blobs bucket_name = accessible_blob.bucket.name # Create the OAuth credentials from the downscoped token and pass a # refresh handler to handle token expiration. We are passing a # refresh_handler instead of a one-time access token/expiry pair. # This will allow testing this on-demand method for getting access tokens. def refresh_handler(request, scopes=None): # Get readonly access tokens to objects with accessible prefix in # the temporarily created bucket. return get_token_from_broker(bucket_name, _OBJECT_PREFIX) creds = credentials.Credentials( None, scopes=["https://www.googleapis.com/auth/cloud-platform"], refresh_handler=refresh_handler, ) # Initialize a Cloud Storage client with the oauth2 credentials. storage_client = storage.Client(credentials=creds) # Test read access succeeds to accessible blob. bucket = storage_client.bucket(bucket_name) blob = bucket.blob(accessible_blob.name) assert blob.download_as_bytes().decode("utf-8") == _ACCESSIBLE_CONTENT # Test write access fails. with pytest.raises(exceptions.Forbidden) as excinfo: blob.upload_from_string("Write operations are not allowed") assert excinfo.match(r"does not have storage.objects.create access") # Test read access fails to inaccessible blob. with pytest.raises(exceptions.Forbidden) as excinfo: bucket.blob(inaccessible_blob.name).download_as_bytes() assert excinfo.match(r"does not have storage.objects.get access") system_tests_sync/__init__.py 0000644 00000000000 15025175721 0012452 0 ustar 00 secrets.tar.enc 0000644 00000024124 15025175721 0007500 0 ustar 00 % �MU[� _���k�f �(|\�@j��=o� M(��P ��o�I�b��X�}}�r���ʱ�[7)�C8�٬�r>���0�3$ДI�w-�N�y�E��1焥�����W �,�j9�E/!5鋇̉��_����=��i�/B��G�Oc˼�w*p�pH^��i11h��9�B�/���җ�R�m�ǧ���w����|�*`~-��P����Հ�(*(��g2pm�+x9�hP�}�.O1��ȣw�x r�oRM^�ob����πIT�Ĺ�@"��"<�l���%�tCG!l�!�Pg� ��M6I�a���"B�Y�܃y�5�ғ�� �: �AM���H���t �w� �@bܮ�Ӻ��J��b���� + ���V��$��`Lj�h�3Jj&[G�� ��Qx�p � �;�K� Nvxq�}��- ����Gv䰿d*!'���ԧ#�Lp� �kf�w~�ܳX�l�! F�"� ~ 8f�0�rpP5�y�E�tMǠJQ�`��,�?҄�=D���4 pףg�,]e#��I����GĈ]�8��r�E�����\ �Nx5'1��$�#W^��ϿF�0!�.V9KבT;7 �.��-��[�!��T�=�Ρ`W�C��A N�`n)���h���(� ~uL�I�nk�Npr= �Z����.�8K[8�p-�Al�]�f�D��^>�~�VE��o�����D��Ә �W�g�����bf�!46Ao��Z��~�@��?80�� 5��д&,���)�I�w���`) �0�C�}�㪗=���� Q<5�?褚��@�YY8*�PᡪE�\�E�>G�yMI�n��5�ق�:VF�!��݊�)� �LM��:�F�>����˙�e{��rJr_2嫹�N?��zlH�� �뾒���(�H�j��}(��P'bt 89�p �r$b����%ف P��ٳ�v�x{{�> �.)G�Ë��T uծ��X�fN��u?�.W ,��!$ #F���V":#���'Q7ep&�ЧN'1�[ Js�3���4'��W��J=u��O�? �#�(`=��a D�"*� �x 0���ݔ�2�X�0b�J2�+�b�����'�� (�f���O�jX��w�HQݚ��x� >�Y�����B� ���G�Х�z�LX(+�0��K���5�$)`��h&{Ic���4]�Ey���)|�j� /�Vo�i��|* �;:�$� ������Hj ����[/��t��1 W���� � Pȹ&� ��� �D\�|��XA%��aO��^�=�o�5���T � X�~�yح��pE�.�h#����d)>��N�{�]A`���0J�n=_oj맋�reoFR��sm�s�`̦���=� �=���a ��<��Z�Z9�e�i�Ս�L I���Q/:�:L;� ���H<~_���V�0<uz�a�R�~"��jS�� � ڙ�(��;O:��Aim�?�KV)�|� t0!����`@�����0�AV��'� s!�jo�q3#�D��V�G����Ðc� Z+�rK&&ڤ�YE5|؆?v��Aя�cd�@��� ���}9�pV���� �B�-2����Ɨڳ�,� ���w���;9q�܍G��U~� %D [i16��� Oa�6tu���G��P�H+�VOp�7j���S|����?��"����D�6���� T#� ��KfAI� r mPxUm�<�����`���m�fiD�ur�K4�V��9*��(����Iڭ}�I�G4`i �|��JYN�u�'���<�^� �v���[A5/)��S�FK�A�1����d��)�h�y��/Oѡ ����Q3&ΐ�l� �#Lu��TX���p�a8'�E<�@�z�{J��� �a��y"{�8TRbQ��#.w��q~8б��Z �J rkc�kÝ����OF�{ti��x�P����E�]� 5�t �2N@#$vM!������Vۮڭ� 0·��lcU>�8Lb�ⲩ������2�hIO@ ���Vo ���O�Dy �GĹ ��N� D7P~���8�� b���v|�T � �Z=O~ �ź�*�ro^�B�`ީ +E�l���WKHHm*]!Ϭ�&��xʺ9(�� Վ�3VE����C��Z����m��D}/���8�Y� �~)�� E�%���i�81tkBh���~�|�M��� |� �4o�N�eR_�>����6l��R A2L�� ��ϲ����]P��z`�f�� h^�����n1�2����C�t���5�:K��OXM5� 9��"( �*C�5���L�wI�V�Y��#�w�Fe �X�Tb ��BЮb6�-�ۘP�J^�f�����.�n(��@Z�4�]{G(�/]��P� �Um_ɷ�١Z�5i�O� 銋���xv� �c������ӾK�F�m�/��h�R� *�j�HU�tM�C���>X�C�c]ާ�q2.P����9��~g#�z �W����my4cr#�*h%kzj�/D VҖ-� �u ��m-�T�4�/;4!���[!�%�!�� ^s��h/a��8L��\���� �� Th4�z,�ι9�"3 ��;��W>�i]�Y=� M�E@���i^ ���A�%���!J�y��F��N�7t��@�n?jT?�����1~���A���oWd��`����f�$e7��G�B�H���AvЧ�x��2��V�8$�AK"�K/%1�T�ټXʷ9)�i��G����Zԍ��J��9� \G �����8 g����Vz2�� ��k ��� ��~)h�U�!@k¼=K6���t��I40�ձۤS4��M o��#���#��8g.#����8}��|W�ʵ���L�e�Avc� V��d�5���jo��K T����b��[4ge�S���ڨ$�<����\��8â8R�ky�6�8��uX��/}C���� ��}<1ߏ�$=�6['� ;ܿ�v�i1v��� <��<��tnMͫV�z�ʹD�|j�R+��8�FNt˓��R.��3����4�-�:�ӣx��Ch�W#�� �`g ;��p���6�ʰ���V��A���l/�g���.c�Ő��Aw���t�����I�T ��0e��� _ŋn`�ޭ�W�_��?�<����7��R��hp�ߔ �+Z�� � c��o c��<7�4��=��"��� �G�W�h����ƪ����s���g�� dׂm�q4�z���-l-ou��(�P��_Ǩ���;�W��� ��3����3�f>��s�/E�&!�9�t� �¹ǻK��8����p?JOÉa���������ܳ � �٠<��x%V�:�3?��} �H��i�܊����+?�(���g¶T:�H����P�M����0$�^ � g>�����.�5�펦G� A5M��U��LQ���ƒͅ�E�Ć���rz/%�R���k �� Ȃ>p�mҾUگ%t���D@��������(��R9�=h�Q�y]�[y^I����n�!�j��k=�+ױ�8*����!f=X�-��Hdq�*��q;hf��g�\ �������Z��|Q(\����$ǘ����ɩ�hhRs�4E�S͚gR������q���KK�P�Io �ځ-����>7����� �F�pZ ǒ����c��'v(y4J�`�S���ig �~W���]0l� �l;Px�%y3|�X%�,/�zͽ�]�s �H]b�i6kY��§{�d Eđ���}��/<�0|�{�)g�V,� ��*]�4=W`y�x��}`C��,#�����c�YE�"b,w���2���9^L�-#�}|d+�=��Z��K9vd-��Ur�;Cd��]�렾{u����!���m��>qZ���� �9n��Ra}4�����OM�f�H_��J����]yT�՚� �1� �L��v~�;"u�W &T�� NyZ�}/:c�~��]0�ZkL4%�N[������Ȓ���0�r�V�G$O�Y���ӉR���\�ݫ �0�q��Ơ2��5�븎Q ���_J}:>E ;c:���(��������FF�W����M^���K�G=�1Q~�r�3l�le��g8�Đs(�gNŁ �TA �Y H|zthՖ >��GkLT@���#x��� ���)����kB� �jl�4�jN���4q>4@<�q��o)�Rq=壵��B�_�6�)����m>�9��"şcXG9����N�EHQ��^���{�T�b�I����� 2�, JW�4hʯ:lU��G�������G����s����"��> b����,�Tb>t�!�f;�L� Ǜ�Df���2 ���J�n.>�`�t�I)���R~E��D'(�� ���iK� �a`��G� �mW�Y�z���h?���g3N CKt$9� ̴��8>�<P��6c�!�eN��j۳ {�54�s�-��<l[� 4�¤��c�X9o�7)��C@��[�����O� �ֆ̀�� ��j��c� eB��b���|��4���F��ʪ���40j�l3������|H&� �cn�~��24.ˌ4Pn��0q�_���M�S,� P'��i3� ��|Sc��F�L��;��vg+u�ڸ'l�@��~MWb�(Fz9}����4�3 �Ԃ�8}2�qu@Y5T�+��X��d}� ��� *DF��t� (�:.g��T�n�C��G+I�6m^��Ι$�Օ}�Q�p�6VO �ʕA+���mc� 1�h�����~c�?.OI��_A���1����9�)� � ��-m\�%�z��ݺR�*=-���x�V5Ȟ�<Oz(%a>�~P������hSvM����3V]�S���O�٥f��g�hW��� j���u��G��7�VӢ>7ȕ;Q^_�RDr�M]�4�К�}�u�?�>��%�e{��i܀ß�����C@���G��V%�{̔��gΗ�!�ߞl ���{Bۋg+)�3(T`�'��f���PS$ p�C�)(�tЩ�ɡ�;kE� ��:t��[�c�eb>�ؒ"�"��9�� �0�>�b���ǧ����$P����6�(т}�!Ց �[�,)Op�t� �K�?�Ȏ�����`hv*���Hڵ[pYP��~�18�=�u ���Fg�U�¥�6��������~8� ��\H�㈱ϣ6��PD�{� T ���Lw���&��ݨ��pmNΨ�іu:�a��3�YM�:���< �ߠµ3b~�N�|/�� �y� ���6td �- �~��ÖmxD�� �,pt�韲ώ0d<T��zm q�IȨ�g�$[���KZ4!R��K�����}��8��A��4S�!��u ��� S�h�"�8��1��aU\e��K�AC�� "�r�Ϙ! g��"bW1�͔"�����;�Y-��ҍ ��� z+��_��Js���u��#=�X��ƁQ瘧�>3_����q]n���Ŝ�6y y"r��f'��W��?;�mH����$ ��]��� �s:ug�YE�⥢V���@8 �Џ�S{�!k��ѧ�ў�>�R3/q����8.�>qn I�r���D��?��*���{�Ad ӓT�4;oK� ����wƻ�6 �*E�p����{m�d�3��va���*���&{#�x���>E�L�oP XT��"�;}L u�.e��tZ��&����2m}]�{^����l��P�H��ri5�{il���K%e ��cDn�֨;u;�Tfθ�2���K �H��7�Źd.����J��5[�(+K�t�& L��9��Iy�� ������;FL�����r린�i�/;N���D3�^�5��7!��U� �"�jA�J�-Ԫ�$�lv�_-̹ǰ������4��\�]f R �{o|�S��(#�`|�Oui� �FGoyWB�|���DD;�]8;�e�eV~6B�������J!ʕ5°n�:A��6�k�Ge �t` ���VSz�g�"Qa�:Dӭ9DTy 8�<|�nS �q �wb� u��J��H��L� ��8Ȥ\��P�F�q�@��;6 M���]'C��9��4��r@���HKc��'��3)��؛�����MBH���Ua� zB�1p��e��v�69��pH5C� �)C��j����Ȉ��, ��\�+5O $f��q�Ҥ� ����|�R@6~Z���b^ m�!��_�"���fX�1�VOc�_�&گ�{ӯ� ���f]�� T퉲�nu!%L��n j.9�ͭ��5� )��ci fT��iZ���C�K� ���#�8 H���j�N�d�: PL��u�fVʂ�X:IY��x�X����\'��08 �k�QxGH W u�����' ���D��+����_I<�d'�ȵh�ϴ�<����o=��eX8�k�d | uD1�$�A��x� �I9)����m���n<p��Ԏ�nW��@Ki� ���{>�" ��#�ge]���j���B��d���r���1��o��NoiJ���3:7?����M(��/e��t߰�8B��+��dE�e4=Ž�S ^0��9@[Aw �U3 v oBu ��i�" g�S ���dņR�k��h�KK|e'kV��D1���3���m 8SzJ��:K,J!BD����*-�8���� 3��zGE�5h6"��K�}BJ�� �0iY��� T��5���[ E�8�)ܡ_�.���ΉA�3I���"��9-�6Q�6��� ڌ;�;o_wѣ���q?�*�eZ@��v���W� ��߈c���m��c=�h�h���5���,�×���]r�X���̣b��� �d�����=g� L�l�N�P�^��Vqƭ6�� �YW�P��� B���qypKm��d�ײ���1I���{>~�?��n�j� ��^\=�A��](����<U]�B<yՂ��i�����Hex�����\v��yHth��e҅�Z�'T�,��� �G=�?�*s3��F\� �E�d�Ijв `�"�ɘj�dGwj���� .�O�J�����Ő�ã��V:Jg3r���t� ��<E��a�`�'{w�����b�W�)>#? � D��s���YH.��]k��&c ����#aZ��>u>]��Ū9P�9x� � $��/sb�%|+ �n}����*�ma�zH��)T���X���K��]Yd~�n �t �}*�x���n �������`�#͘^�m[w��|B�*6E��GH��.�,���M �K�p��n!7pgl@E�l���Y)�PG��G����3_ XWc�fq�(�x_ ��t|}���[DV���`Y�6�6n?%��8x*͉n�U���M ��S$w<�J�*�g��?jgu�� ��mr &{�w���8�Z�d�V���cr�Z�F���2��3K Й�$ׁ�� �on� I��~T?�*�� zς��r����85�����f�k�)�^@�^NJ �_�3 ��O����M*xN=L� . � ��_��=&��JK���h;mʭ+r��4�53�s��� �v\�JOz�p�X ����<7�4V��t +x�}�h�o3�+O'���źm�˔�\����n�>z � A���/���p��/� �G�zᒯ�g��m��s��|�� ��;D�qߚ�M ���[���X��n��l���'VDm�h�� �l GSvE��1����뷇����� �|���Ф���O 7�Ꝉ� ,z�|���4whI�� �� ������&�#&����^��� ���@Df��<~Y�x�wp�X�?�E.�8B1�l�#[�+��k��8Aʑ��֤}�d�sΈ�B�Z-۹�Hh#+6$�g�ʗ�D]��S22n�=3Q WUO���>�[;�!\�t��>�+�����5��E��4��2�kp:58��1j ��遲*Ʒ���!�;���O���jN���0F=�w�*�s �Z�J#�\���=#����P���<�9Y�q�:-��$���1�xp#{2HDˉD�Ϙ�;�n�vݺ 9��&b�Q�|��Ҝ%�_ �T�w�}������l��l������Ԍ�ai�'��]I��dI>�&䑤��UNj�<V~� #5/c��F_2ky��^"���W%k7vu��K��Ա��,�� � ��2he"[�6 �Q�/� �ʬn��r=}�� ��Ri~ �ih��#�In���d�� o��Yr�ˮ�o�ܼ�<��J��v}����<d���2�|�y4��U;��Q���8�Z<a����7� mg�[I)gO^o����\C�or���)�-^F�'�so��շx�.�{X��\R) �Gs~A���cw[�B�XD�[=+�̾k������5ݣ�.,){=ڻmO��nV����ҷ"��� �8�l .<��œ?Q�^��1� ��Ma����><.?����W�]�?V5-R~H��I˰W��3���WF5�7�{r��0܅W�|� � S,O������|D ���/��EG2��{#71�}N��'�e�ڭ��j�@�G�=G�0�� V�Ul�余m�7�*�X��QaQ�u�kO�+�|�n��� �(�眏 4T̘���g� �� �ݳk���Z��9Y�My�1�\��J���Q�<�x,��*> ���c��L_X�ծfe������7��Xy�&6��@0��u�����i ���}��O7jҦ�Ã=�̲Ցt &u'g�9��ڗR��Z�W������1o�4�w�)��g�_�S9" �Q�3���΅:�O�&����)<�vPH4�rPD>Dyo�I W/ �< ��zp\�vgj�]7���� �Ml���d�4��M�m���e� �D\І�=a����f@/N{������[�M?/�4#�Ro������Z�@�xtO�f��T�C���} �L� r�)c�7i��r��="�'R�(W&�s9�.��z�W�YuwX�r�[����>@�)��͞o� ��υy~�6���k�,� �֦R&���!O�/��8�v�E-ŸޏL7�O�q4R�J���%h�Zq��d`�(��~� �d���p��lVMr{�T�r��)�ʨ!��e}W-<>�ԍ)]� \��ɲ� d�rr��w^�tZ"�V=� O&D�5�[r�U�!9;�mdl�?J>�b"t9�jIm 9P)��i��P>��9?+��ڂ�D � � oط H8�t� !�m��G0�}'@���a���r�S��dN��y%�@�Y-%���Íf�3>.���_�c���b9p*��@!�F��k��5t~tv�@��m�p^��̡L�XD kbz�XC �SJs�2 �JI��*�/VG�э���{��R^�2{tܔ�b��$��ƨ��![N�pRf�'�A�y��} v�i��FB"-9���+�?xf�> z �LC-8@o8����4�Dd �L���P�7�U�>�vt�1�IL�ixN�dVއ*� Be3���d�ٯQ�dK2�io�?s�M�:�H=?�Ɉ L[���BL�*����ˣ�UA��җs(��:�>� ���sV�$�O�C�����rHY�ޠ�t�ƗS�p��ހV]�x�l��D*�u���vO���O�����A�<E�&� ��6 �������f���\�c9B����L�5� ]Ov�.��C�(�accF{C�E#��O��3��-�ـ��K̍[Cx[ϥN��^��3�u���E |#�w7�n��0�|��l�Ƭ����K��B��$���ޮ_�����R{ค��{ F�eT�4 R�x+�+ Rq��¸Ws"��l�i�:8�C��_�(J"��4�]Qea�[Ei.�����͆�x�%Z���Λ,����d���xq�r�f�`��ަ ��tM^�B!�_��ώL[_ �Z�!r;�Bh�]�b���X�|$ �l����S2�Z��Hޘ��Lg�o�v=ˌ&}%�ȟ�����a���B �����M7Ū��t��K ;>@���D�U� ��5� /�ǽE ����o�k��N��K�2���3m=��Qk���It*���q��~Cy�]%\�`H GN0h��G���|�褃T٣�?��zND� noxfile.py 0000644 00000034140 15025175721 0006571 0 ustar 00 # Copyright 2020 Google LLC # # 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. """Noxfile for automating system tests. This file handles setting up environments needed by the system tests. This separates the tests from their environment configuration. See the `nox docs`_ for details on how this file works: .. _nox docs: http://nox.readthedocs.io/en/latest/ """ import os import pathlib import shutil import tempfile import nox HERE = os.path.abspath(os.path.dirname(__file__)) LIBRARY_DIR = os.path.abspath(os.path.dirname(HERE)) DATA_DIR = os.path.join(HERE, "data") SERVICE_ACCOUNT_FILE = os.path.join(DATA_DIR, "service_account.json") AUTHORIZED_USER_FILE = os.path.join(DATA_DIR, "authorized_user.json") EXPLICIT_CREDENTIALS_ENV = "GOOGLE_APPLICATION_CREDENTIALS" EXPLICIT_PROJECT_ENV = "GOOGLE_CLOUD_PROJECT" EXPECT_PROJECT_ENV = "EXPECT_PROJECT_ID" ALLOW_PLUGGABLE_ENV = "GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES" # The download location for the Cloud SDK CLOUD_SDK_DIST_FILENAME = "google-cloud-sdk.tar.gz" CLOUD_SDK_DOWNLOAD_URL = "https://dl.google.com/dl/cloudsdk/release/{}".format( CLOUD_SDK_DIST_FILENAME ) # This environment variable is recognized by the Cloud SDK and overrides # the location of the SDK's configuration files (which is usually at # ${HOME}/.config). CLOUD_SDK_CONFIG_ENV = "CLOUDSDK_CONFIG" # If set, this is where the environment setup will install the Cloud SDK. # If unset, it will download the SDK to a temporary directory. CLOUD_SDK_ROOT = os.environ.get("CLOUD_SDK_ROOT") if CLOUD_SDK_ROOT is not None: CLOUD_SDK_ROOT = pathlib.Path(CLOUD_SDK_ROOT) if not CLOUD_SDK_ROOT.exists() or not CLOUD_SDK_ROOT.is_dir(): print("{} did not exist! Please set the CLOUD_SDK_ROOT environment variable to a directory that exists".format(CLOUD_SDK_ROOT)) exit(1) else: CLOUD_SDK_ROOT = pathlib.Path(tempfile.mkdtemp()) # The full path the cloud sdk install directory CLOUD_SDK_INSTALL_DIR = CLOUD_SDK_ROOT.joinpath("google-cloud-sdk") # The full path to the gcloud cli executable. GCLOUD = str(CLOUD_SDK_INSTALL_DIR.joinpath("bin", "gcloud")) # Cloud SDK helpers def install_cloud_sdk(session): """Downloads and installs the Google Cloud SDK.""" # Configure environment variables needed by the SDK. # This sets the config root to the tests' config root. This prevents # our tests from clobbering a developer's configuration when running # these tests locally. session.env[CLOUD_SDK_CONFIG_ENV] = str(CLOUD_SDK_ROOT) # This set the $PATH for the subprocesses so they can find the gcloud # executable. session.env["PATH"] = ( str(CLOUD_SDK_INSTALL_DIR.joinpath("bin")) + os.pathsep + os.environ["PATH"] ) # If gcloud cli executable already exists, just update it. if pathlib.Path(GCLOUD).exists(): session.run(GCLOUD, "components", "update", "-q") return tar_path = CLOUD_SDK_ROOT.joinpath(CLOUD_SDK_DIST_FILENAME) # Download the release. session.run("wget", CLOUD_SDK_DOWNLOAD_URL, "-O", str(tar_path), silent=True) # Extract the release. session.run("tar", "xzf", str(tar_path), "-C", str(CLOUD_SDK_ROOT)) tar_path.unlink() # Run the install script. session.run( str(CLOUD_SDK_INSTALL_DIR.joinpath("install.sh")), "--usage-reporting", "false", "--path-update", "false", "--command-completion", "false", silent=True, ) def copy_credentials(credentials_path): """Copies credentials into the SDK root as the application default credentials.""" dest = CLOUD_SDK_ROOT.joinpath("application_default_credentials.json") if dest.exists(): dest.unlink() shutil.copyfile(pathlib.Path(credentials_path), dest) def configure_cloud_sdk(session, application_default_credentials, project=False): """Installs and configures the Cloud SDK with the given application default credentials. If project is True, then a project will be set in the active config. If it is false, this will ensure no project is set. """ install_cloud_sdk(session) # Setup the service account as the default user account. This is # needed for the project ID detection to work. Note that this doesn't # change the application default credentials file, which is user # credentials instead of service account credentials sometimes. session.run( GCLOUD, "auth", "activate-service-account", "--key-file", SERVICE_ACCOUNT_FILE ) if project: session.run(GCLOUD, "config", "set", "project", "example-project") else: session.run(GCLOUD, "config", "unset", "project") # Copy the credentials file to the config root. This is needed because # unfortunately gcloud doesn't provide a clean way to tell it to use # a particular set of credentials. However, this does verify that gcloud # also considers the credentials valid by calling application-default # print-access-token session.run(copy_credentials, application_default_credentials) # Calling this forces the Cloud SDK to read the credentials we just wrote # and obtain a new access token with those credentials. This validates # that our credentials matches the format expected by gcloud. # Silent is set to True to prevent leaking secrets in test logs. session.run( GCLOUD, "auth", "application-default", "print-access-token", silent=True ) # Test sesssions TEST_DEPENDENCIES_ASYNC = ["aiohttp", "pytest-asyncio", "nest-asyncio", "mock"] TEST_DEPENDENCIES_SYNC = ["pytest", "requests", "mock"] PYTHON_VERSIONS_ASYNC = ["3.7"] PYTHON_VERSIONS_SYNC = ["3.7"] def default(session, *test_paths): # replace 'session._runner.friendly_name' with # session.name once nox has released a new version # https://github.com/theacodes/nox/pull/386 sponge_log = f"--junitxml=system_{str(session._runner.friendly_name)}_sponge_log.xml" session.run( "pytest", sponge_log, *test_paths, ) @nox.session(python=PYTHON_VERSIONS_SYNC) def service_account_sync(session): session.install(*TEST_DEPENDENCIES_SYNC) session.install(LIBRARY_DIR) default( session, "system_tests_sync/test_service_account.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_SYNC) def default_explicit_service_account(session): session.env[EXPLICIT_CREDENTIALS_ENV] = SERVICE_ACCOUNT_FILE session.env[EXPECT_PROJECT_ENV] = "1" session.install(*TEST_DEPENDENCIES_SYNC) session.install(LIBRARY_DIR) default( session, "system_tests_sync/test_default.py", "system_tests_sync/test_id_token.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_SYNC) def default_explicit_authorized_user(session): session.env[EXPLICIT_CREDENTIALS_ENV] = AUTHORIZED_USER_FILE session.install(*TEST_DEPENDENCIES_SYNC) session.install(LIBRARY_DIR) default( session, "system_tests_sync/test_default.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_SYNC) def default_explicit_authorized_user_explicit_project(session): session.env[EXPLICIT_CREDENTIALS_ENV] = AUTHORIZED_USER_FILE session.env[EXPLICIT_PROJECT_ENV] = "example-project" session.env[EXPECT_PROJECT_ENV] = "1" session.install(*TEST_DEPENDENCIES_SYNC) session.install(LIBRARY_DIR) default( session, "system_tests_sync/test_default.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_SYNC) def default_cloud_sdk_service_account(session): configure_cloud_sdk(session, SERVICE_ACCOUNT_FILE) session.env[EXPECT_PROJECT_ENV] = "1" session.install(*TEST_DEPENDENCIES_SYNC) session.install(LIBRARY_DIR) default( session, "system_tests_sync/test_default.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_SYNC) def default_cloud_sdk_authorized_user(session): configure_cloud_sdk(session, AUTHORIZED_USER_FILE) session.install(*TEST_DEPENDENCIES_SYNC) session.install(LIBRARY_DIR) default( session, "system_tests_sync/test_default.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_SYNC) def default_cloud_sdk_authorized_user_configured_project(session): configure_cloud_sdk(session, AUTHORIZED_USER_FILE, project=True) session.env[EXPECT_PROJECT_ENV] = "1" session.install(*TEST_DEPENDENCIES_SYNC) session.install(LIBRARY_DIR) default( session, "system_tests_sync/test_default.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_SYNC) def compute_engine(session): session.install(*TEST_DEPENDENCIES_SYNC) # unset Application Default Credentials so # credentials are detected from environment del session.virtualenv.env["GOOGLE_APPLICATION_CREDENTIALS"] session.install(LIBRARY_DIR) default( session, "system_tests_sync/test_compute_engine.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_SYNC) def grpc(session): session.install(LIBRARY_DIR) session.install("six") session.install(*TEST_DEPENDENCIES_SYNC, "google-cloud-pubsub==1.7.2") session.env[EXPLICIT_CREDENTIALS_ENV] = SERVICE_ACCOUNT_FILE default( session, "system_tests_sync/test_grpc.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_SYNC) def requests(session): session.install(LIBRARY_DIR) session.install(*TEST_DEPENDENCIES_SYNC) session.env[EXPLICIT_CREDENTIALS_ENV] = SERVICE_ACCOUNT_FILE default( session, "system_tests_sync/test_requests.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_SYNC) def urllib3(session): session.install(LIBRARY_DIR) session.install(*TEST_DEPENDENCIES_SYNC) session.env[EXPLICIT_CREDENTIALS_ENV] = SERVICE_ACCOUNT_FILE default( session, "system_tests_sync/test_urllib3.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_SYNC) def mtls_http(session): session.install(LIBRARY_DIR) session.install(*TEST_DEPENDENCIES_SYNC, "pyopenssl") session.env[EXPLICIT_CREDENTIALS_ENV] = SERVICE_ACCOUNT_FILE default( session, "system_tests_sync/test_mtls_http.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_ASYNC) def external_accounts(session): session.env[ALLOW_PLUGGABLE_ENV] = "1" session.install( *TEST_DEPENDENCIES_ASYNC, LIBRARY_DIR, "google-api-python-client", ) default( session, "system_tests_sync/test_external_accounts.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_SYNC) def downscoping(session): session.install( *TEST_DEPENDENCIES_SYNC, LIBRARY_DIR, "google-cloud-storage", ) default( session, "system_tests_sync/test_downscoping.py", *session.posargs, ) # ASYNC SYSTEM TESTS @nox.session(python=PYTHON_VERSIONS_ASYNC) def service_account_async(session): session.install(*(TEST_DEPENDENCIES_SYNC + TEST_DEPENDENCIES_ASYNC)) session.install(LIBRARY_DIR) default( session, "system_tests_async/test_service_account.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_ASYNC) def default_explicit_service_account_async(session): session.env[EXPLICIT_CREDENTIALS_ENV] = SERVICE_ACCOUNT_FILE session.env[EXPECT_PROJECT_ENV] = "1" session.install(*(TEST_DEPENDENCIES_SYNC + TEST_DEPENDENCIES_ASYNC)) session.install(LIBRARY_DIR) default( session, "system_tests_async/test_default.py", "system_tests_async/test_id_token.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_ASYNC) def default_explicit_authorized_user_async(session): session.env[EXPLICIT_CREDENTIALS_ENV] = AUTHORIZED_USER_FILE session.install(*(TEST_DEPENDENCIES_SYNC + TEST_DEPENDENCIES_ASYNC)) session.install(LIBRARY_DIR) default( session, "system_tests_async/test_default.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_ASYNC) def default_explicit_authorized_user_explicit_project_async(session): session.env[EXPLICIT_CREDENTIALS_ENV] = AUTHORIZED_USER_FILE session.env[EXPLICIT_PROJECT_ENV] = "example-project" session.env[EXPECT_PROJECT_ENV] = "1" session.install(*(TEST_DEPENDENCIES_SYNC + TEST_DEPENDENCIES_ASYNC)) session.install(LIBRARY_DIR) default( session, "system_tests_async/test_default.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_ASYNC) def default_cloud_sdk_service_account_async(session): configure_cloud_sdk(session, SERVICE_ACCOUNT_FILE) session.env[EXPECT_PROJECT_ENV] = "1" session.install(*(TEST_DEPENDENCIES_SYNC + TEST_DEPENDENCIES_ASYNC)) session.install(LIBRARY_DIR) default( session, "system_tests_async/test_default.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_ASYNC) def default_cloud_sdk_authorized_user_async(session): configure_cloud_sdk(session, AUTHORIZED_USER_FILE) session.install(*(TEST_DEPENDENCIES_SYNC + TEST_DEPENDENCIES_ASYNC)) session.install(LIBRARY_DIR) default( session, "system_tests_async/test_default.py", *session.posargs, ) @nox.session(python=PYTHON_VERSIONS_ASYNC) def default_cloud_sdk_authorized_user_configured_project_async(session): configure_cloud_sdk(session, AUTHORIZED_USER_FILE, project=True) session.env[EXPECT_PROJECT_ENV] = "1" session.install(*(TEST_DEPENDENCIES_SYNC + TEST_DEPENDENCIES_ASYNC)) session.install(LIBRARY_DIR) default( session, "system_tests_async/test_default.py", *session.posargs, ) __init__.py 0000644 00000000000 15025175721 0006650 0 ustar 00 system_tests_async/test_default.py 0000644 00000001700 15025175721 0013547 0 ustar 00 # Copyright 2020 Google LLC # # 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. import os import pytest from google.auth import _default_async EXPECT_PROJECT_ID = os.environ.get("EXPECT_PROJECT_ID") @pytest.mark.asyncio async def test_application_default_credentials(verify_refresh): credentials, project_id = _default_async.default_async() if EXPECT_PROJECT_ID is not None: assert project_id is not None await verify_refresh(credentials) system_tests_async/test_id_token.py 0000644 00000001664 15025175721 0013730 0 ustar 00 # Copyright 2020 Google LLC # # 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. import pytest from google.auth import jwt import google.oauth2._id_token_async @pytest.mark.asyncio async def test_fetch_id_token(http_request): audience = "https://pubsub.googleapis.com" token = await google.oauth2._id_token_async.fetch_id_token(http_request, audience) _, payload, _, _ = jwt._unverified_decode(token) assert payload["aud"] == audience system_tests_async/test_service_account.py 0000644 00000003355 15025175721 0015307 0 ustar 00 # Copyright 2020 Google LLC # # 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. import pytest from google.auth import _helpers from google.auth import exceptions from google.auth import iam from google.oauth2 import _service_account_async @pytest.fixture def credentials(service_account_file): yield _service_account_async.Credentials.from_service_account_file(service_account_file) @pytest.mark.asyncio async def test_refresh_no_scopes(http_request, credentials): """ We expect the http request to refresh credentials without scopes provided to throw an error. """ with pytest.raises(exceptions.RefreshError): await credentials.refresh(http_request) @pytest.mark.asyncio async def test_refresh_success(http_request, credentials, token_info): credentials = credentials.with_scopes(["email", "profile"]) await credentials.refresh(http_request) assert credentials.token info = await token_info(credentials.token) assert info["email"] == credentials.service_account_email info_scopes = _helpers.string_to_scopes(info["scope"]) assert set(info_scopes) == set( [ "https://www.googleapis.com/auth/userinfo.email", "https://www.googleapis.com/auth/userinfo.profile", ] ) system_tests_async/conftest.py 0000644 00000006560 15025175721 0012722 0 ustar 00 # Copyright 2020 Google LLC # # 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. import json import os from google.auth import _helpers import google.auth.transport.requests import google.auth.transport.urllib3 import pytest import pytest_asyncio import requests import urllib3 import aiohttp from google.auth.transport import _aiohttp_requests as aiohttp_requests from system_tests.system_tests_sync import conftest as sync_conftest TOKEN_INFO_URL = "https://www.googleapis.com/oauth2/v3/tokeninfo" @pytest_asyncio.fixture def service_account_file(): """The full path to a valid service account key file.""" yield sync_conftest.SERVICE_ACCOUNT_FILE @pytest_asyncio.fixture def impersonated_service_account_file(): """The full path to a valid service account key file.""" yield sync_conftest.IMPERSONATED_SERVICE_ACCOUNT_FILE @pytest_asyncio.fixture def authorized_user_file(): """The full path to a valid authorized user file.""" yield sync_conftest.AUTHORIZED_USER_FILE @pytest_asyncio.fixture async def aiohttp_session(): async with aiohttp.ClientSession(auto_decompress=False) as session: yield session @pytest_asyncio.fixture(params=["aiohttp"]) async def http_request(request, aiohttp_session): """A transport.request object.""" yield aiohttp_requests.Request(aiohttp_session) @pytest_asyncio.fixture async def token_info(http_request): """Returns a function that obtains OAuth2 token info.""" async def _token_info(access_token=None, id_token=None): query_params = {} if access_token is not None: query_params["access_token"] = access_token elif id_token is not None: query_params["id_token"] = id_token else: raise ValueError("No token specified.") url = _helpers.update_query(sync_conftest.TOKEN_INFO_URL, query_params) response = await http_request(url=url, method="GET") data = await response.content() return json.loads(data.decode("utf-8")) yield _token_info @pytest_asyncio.fixture async def verify_refresh(http_request): """Returns a function that verifies that credentials can be refreshed.""" async def _verify_refresh(credentials): if credentials.requires_scopes: credentials = credentials.with_scopes(["email", "profile"]) await credentials.refresh(http_request) assert credentials.token assert credentials.valid yield _verify_refresh def verify_environment(): """Checks to make sure that requisite data files are available.""" if not os.path.isdir(sync_conftest.DATA_DIR): raise EnvironmentError( "In order to run system tests, test data must exist in " "system_tests/data. See CONTRIBUTING.rst for details." ) def pytest_configure(config): """Pytest hook that runs before Pytest collects any tests.""" verify_environment() system_tests_async/__init__.py 0000644 00000000000 15025175721 0012613 0 ustar 00
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.02 |
proxy
|
phpinfo
|
Settings