Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-04-14 17:36
📄 .gitignore data secrets.tar	16 B	2025-04-14 17:36
📄 __init__.py	0 B	2025-04-14 17:36
📄 conftest.py # 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()	4.04 KB	2025-04-14 17:36
📄 secrets.tar.enc $��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�� )��?�4Q:�at��{��e,aS�+`��]��s�g�Aܲ��>�A�rT��c�L��u�\=kE��v��sf�V3��k8؇ҸL} ��ػ8G(Qjl�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��G؅qO��=�ء�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��&��7A�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�?��.Ed��=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羐��β��ǌ�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�WRE�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��ԝdc901��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��aSY��pb�o�/M�ෟO�3Y��+S)垵��\|>�p}`�m�B\|��%��ET��So9��[��5~P��b��ڪ��W�RKσ.��Y#��:5��\|�<ɳ�~PO��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%}:ǉ<�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]�94�J7�X��L�R%��8JSF1��+�T��ZNC֎��aX�N\vYt�y�Y뚝S��ɔ<h��;?TS!%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<GM#P��eF��u�8 ��ԎiV��g�7�(��y��n��c ��ր��n�vlI�;��w��J��Þ��!��.o2��X1��K �ym�}�ew"�J1�ί�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�7w1Or��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��Nxgb{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�Ǳ����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��[Wn��]��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��?ʟRY�j�t�."��Z-N�N�grLU��g�\�m�v"~Wo�"��+��q,��gT�E�iR�1�:Z"�i�&�ǲ̜�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��Fd޺fYF�!�@� �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	10.08 KB	2025-04-14 17:36
📄 test_compute_engine.py # 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	2.32 KB	2025-04-14 17:36
📄 test_default.py # 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)	885 B	2025-04-14 17:36
📄 test_downscoping.py # 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")	6.54 KB	2025-04-14 17:36
📄 test_external_accounts.py # 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, } }, }, )	16.11 KB	2025-04-14 17:36
📄 test_grpc.py # 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)	3.43 KB	2025-04-14 17:36
📄 test_id_token.py # 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	902 B	2025-04-14 17:36
📄 test_impersonated_credentials.py # 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	3.37 KB	2025-04-14 17:36
📄 test_mtls_http.py # 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	4.32 KB	2025-04-14 17:36
📄 test_oauth2_credentials.py # 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	1.99 KB	2025-04-14 17:36
📄 test_requests.py # 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	1.57 KB	2025-04-14 17:36
📄 test_service_account.py # 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)	1.91 KB	2025-04-14 17:36
📄 test_urllib3.py # 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	1.56 KB	2025-04-14 17:36

Terminal