# ------------------------------------ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. # ------------------------------------ import asyncio import functools import logging import json from azure.core.exceptions import ResourceExistsError from azure_devtools.scenario_tests import RecordingProcessor from azure.keyvault.certificates import ( AdministratorContact, ApiVersion, CertificateContact, CertificatePolicyAction, CertificatePolicy, KeyType, KeyCurveName, KeyUsageType, CertificateContentType, LifetimeAction, CertificateIssuer, IssuerProperties, parse_key_vault_certificate_id ) from azure.keyvault.certificates.aio import CertificateClient from devtools_testutils import ResourceGroupPreparer, KeyVaultPreparer import pytest from _shared.preparer_async import KeyVaultClientPreparer as _KeyVaultClientPreparer from _shared.test_case_async import KeyVaultTestCase # pre-apply the client_cls positional argument so it needn't be explicitly passed below KeyVaultClientPreparer = functools.partial(_KeyVaultClientPreparer, CertificateClient) class RetryAfterReplacer(RecordingProcessor): """Replace the retry after wait time in the replay process to 0.""" def process_response(self, response): if "retry-after" in response["headers"]: response["headers"]["retry-after"] = "0" return response # used for logging tests class MockHandler(logging.Handler): def __init__(self): super(MockHandler, self).__init__() self.messages = [] def emit(self, record): self.messages.append(record) class CertificateClientTests(KeyVaultTestCase): CERT_CONTENT_PASSWORD_ENODED = b"0\x82\t;\x02\x01\x030\x82\x08\xf7\x06\t*\x86H\x86\xf7\r\x01\x07\x01\xa0\x82\x08\xe8\x04\x82\x08\xe40\x82\x08\xe00\x82\x06\t\x06\t*\x86H\x86\xf7\r\x01\x07\x01\xa0\x82\x05\xfa\x04\x82\x05\xf60\x82\x05\xf20\x82\x05\xee\x06\x0b*\x86H\x86\xf7\r\x01\x0c\n\x01\x02\xa0\x82\x04\xfe0\x82\x04\xfa0\x1c\x06\n*\x86H\x86\xf7\r\x01\x0c\x01\x030\x0e\x04\x08\xf5\xe5\x81\xfd\xa4\xe19\xf0\x02\x02\x07\xd0\x04\x82\x04\xd8.\xb2>H\n\xee\xd9\xd0YE\x04e%\x8e\xd7Cr\xde.F\xa1\xd8W\x11Gw@L;!ght \r\xa8\x06\xb9\x10!\xdb\x0b\xc8\x00\x16g}\xaaa\x1dj\x91lK\x1e\x7f@\xa9x.\xdb\xb0\x04l\xe97\xe7\xeaHM\x96\xa2\xcb\xad\xd8`\x19$\xa5\x1f\xa9\r\xd9\xe0f\xdd}gC\xd6\xacl\x07\x12\xaes\xe8\x11\xd2\xd8b\xf2\xc8\xdf\x12H\xe0\x9bw0\xb3\xed\xb9c\xdf\xee\xc8e\x8a\x0c\x8f\x85\x8e>\x03\xa6\xfe\xd4:S\x8e\x12\x15g\xa4\xe3\xa407l\xde\x03\x88\xbd\xee\xfe\xdf\xb4\xd3g\xb3n\xe6\xb3\x9d\xa3\xa9\xf8N\xbd0=s\xfc2}\x92\x80c\x86\x8a%\xf6\x18Rl\x9c*9\xe7F]5\xdaWR\xdaS\xa4\x01!m\xfa[\xb8@&\xbb\xd8\x86:x\xfbQ\xb9\xd3\xc2\xbel\xd1\xbfjd-\x84\xba\xcfw\x08\xee\x89\x93\xf2q\xcf\xdc<\xa64\xea\x8blZ\xab\xe4\xed\x8c\xd5\x96\x1a,.\xb7C|m\xdd\xe5om\xc3\xe1\xdc\xdd<\x0fXG\x92\x1c\xff(4\xef\x91\x10\x10\xa6\xfa\xd6\xf0\x84\x8a\x9a\x00\xdd\x9b3y\xe4\xf7\xb9\xe7\x11\xdfIa\x81\xee\x03\xf0\xf2\xc6^k\x9e\xc8\xc4\\\xd6\x1d2\xb6\xca\xf4\xec\x96\x8a\x16\xa2\x8b&\x1b\x16\xa7a\x8d\x88\x1b\xf9\xe8\xdcF\xcf9`\xca\x8c\xf6x\x8aV\t{\x92I\xda)\xa6\x97\x13\xf3\xfbg\xb6\x10\xe0\x8a\xa42>\xed\xfc\xd0q\x1c\xf7=7w\x04\xaf\x9b\xb9\xd6|iu\xfcio\xe5:\x02\x92\xf1i\xb1f\x82\xa78\x90MY\xe4\xcdY\x01n\x82i-]\xf7O\x1c\x07q2\x18\xd4^\xa7\x86A\xdf0N\xf6x\x134\r5\xa7\xe8\xbf\t\x08\xec\x85\x7fe\x8a\x1a\xfb\xe4F\xa1\xf5Q\xdd\x96\xd1J M\x17\xa4\xc3\x8f\xfa\x97\x16\xdd07\xf0\x90\x9e\xc1\x80\x99\x00\x066#~\x0f\x89\x98\xee-\xb9v\xd4\xee\xfc\x97;;\x12\xdd\x84\x05\x05\xa4|\x89\xa7*\xd8X\xb7\xef:$\xb9Y\x80^\x101\xe4\x88\xf5\x1a\xff\xc7\x99H\xf071u\x99GTb\xb8;\xee6\xa3#r\xddRK\x07W\x004\xed\x17\xaf%\xfdD\xb5\x92\xc5:\xe7\xbf\x97H/\xba\x97-@\xfe\xeas\xf9~\xf5\xf8.\x07\xa3\xa5\xb4\xef\x9dc\xe5\x93\x13\xeb\x12\xa3\x1a\x1eiy\xee\xccV\xe7n\xc4\x8c\xd7\x8db2\xdd\x84\x9d\xd1\xf2\x13\xddM\x00\xe4\xd2\xc4\xbc\x9fk~Lz&!\xe3D\xbczW[j\xb2\xbbS\xe8\x1b\x06\xb6`\x90GU\x02$\xf2\xea\xb0\xa5C\xbc\x02\r\xc7w\x0f\x03\xf0\x86\xaa\xbeN_`FfP\"\x84i\x8d\xea~\xe0\xbf\xcc8;I4,\xf4\xc0{\x96\x1e~\x05\xcd\xdeoi\x13\xce\xbb7}F\xb4uYh\x9f\xd4V\x00\xcda-\xa3\xba\xc7\x9d\xe2\xbc;\xe9\x95\x8d\xe3V\xa4\xc7d\r\xd0\x94\x9e0\x9a\x87^\xa5s\xe8\x02\x9f\xcf\xc2\x02K\xf7E\x9cA\xb2\x04\xdaW\x88\xc4q\xad\x8f\xd0<\xa8\xbf\xc0\xe3p\xaa\xc6\xc3\xc5\x15\xbb\xbd\x94U*\xce\xfc\xa4\x19\x04\xd2K\x1aJ\x19Y\x93\x91\xa4y\xac\x83X/\xfb\x1e/\xcd\xa9Am\"Z\n\xf5pw\xa5\xa2\xf1\xa3P\xc6\xbb\x9a\xaah]\xf8\x8d\x97d\xb79\x17\xa7K\x99\xaa\x9a~\x15\xf2\x99j*/2;|\x17\xbc\x87\x08\xf9>-\x8aQ\xb1M\x82\xc9\xcfCV\x80\xc0\xea\xb2 \x7f\xeb\x84?\x88\xe9\xa6\x07\xa1\xb3\x1c\x93\xd2RGk\x1d\xad\xf3\xafQ\xda6\x1d\xb1|\x18Qx\xe0\xc0r\x15\xd2\xfa#\xed\xb2X[7\x91\xfdE\xd1r\xf0o\xd6\xdb\x7fm\x8c;\xb59\x88\xc1\x0f'b\x06\xac\xc1\x9f\xc1\xc6\xd44\xa3\xd4\xf8\xdc\xd2G\x7f\xf3gxeM7\xd3\xc2\x85L-\xf2\x19\xc4ZwA\xa7\x10}\x0e\x8bx\x84'\xd1\xdb\xae%\x1b}S\x1b\\\xd1\xce\x17\xe3$\xb5h\x83V\xac\xe7tc\n\x9a\xe2Ru\xf4\xc1*\xf1\x85\xbd\xe8\xc0YS\xb9\x13\x89\xa0.\xfa\x1a2f\xdc\x85\xcd\xc1;\xbb\x0bz\xb6\x87\x9c\x93y\x86\xf3\x01h\xb7\x10#\x7f\r\xf3\xa9\x94}4|\x00\xfe\x80'\xd76\x93\x9dx)\xa0\xcbrY\xb8\xcf\xa2|t\xcc\xfa\xd2u\x1e\xa3\x90\xf7`==\x1b\xa0Z\xbcQ\xf1J\xf2|0]\x0b\xbb\x9c\xce\x171\x1e<4E\x9b\xd9\x87\xf1m\r\xfe\xc1e!\xa6\x1f\x0f\xf1\x96S\xfc8\xe2d.r6\x81\x93\xdeX\xb6\xa3\x86D\x88\xf9\xf2\xd1\x83Z\xbf\"Q\xd1\xf0i\x82\x86\xa9M\xb8\xccg\x91i\xefC\x84U\xcf\xcd\x9b!WVF\xb0\x14\x05E\xaa\x18\x93\"\xc0\xc1\xd2V!t\xe2\xf9\xcd\xfba\xa0\xbc\x15\x14\x84\x9esfK\xbfC\xa2\xedJspo+\x81\x18(\x00\xf6+\x18\xedQ\xe6\xebW^\xf8\x80=\x10\xfb\xd6.'A\x979;)\x06\xf0\x85w\x95S\xd9\x1c9\xcc3k\x03\xf2w\x17\x97\xcc\nN0;0\x1f0\x07\x06\x05+\x0e\x03\x02\x1a\x04\x14\xb1\x82\x1d\xb1\xc8_\xbc\xf1^/\x01\xf7\xc1\x99\x95\xef\xf1A\xf4\x0c\xa1\xde\xd3rj\x85\xbc\xdcW3c\x9d\x137\x17\x80\x01\xb2f\xccFf\x14\xbd\xf9\xbe\xbf\xe1\x9eF\xe7\xaeq\xde\x8c\xa9\x94\x89I2\xb4\xacY\xb4\xd1\x02?\xce\xe0\xf9\x9d\xe7\x8a\xf0\xf1\x8d\xc0\xd5\x8d\x91\xe9P\xcd\xe9\xd4X\x00\xaa\xcdI\t-\xede\xcbT?\xfd\x8b\xa1\xaci0sw(\x1b\xb1\x833\x9fi\x82:9\xb4/ H\x07b\xd3\xf5-\xcbS`\x82\x10\x0b\xd1\x8f\xb8I\xce/\x14\xa20i\xa7\xef\xb8=\xe2Z\x15?z\xa9\x1b\xc2k\x0et\xf1\x18\x16\x07\xd8\x9a\xfd\xea\xe9\xb2Fq\x96\x04r\xcb\x16\xb3v\xfd\xac\xb5*\x07s1\x97\xc9\xe1\xf9I\x18\xe2\xf0{\xce\xa6\xba/\xcd\xf8?\xd2*\x8c\xb3f\xe8\x99h\xa8\x13\x03\xafs,\xa5A\xb3b\x9c#\xb7\xa1\x1b2w\xcd`\xd2\x95\r\xcd\x86\x8aq\xfb?\xfaO\xbe\x9d\xe9B\x9e\x80\xa1\xc2\xb5\\\xad\xf8\xf4i\n\xc8\x80j\x8c\x1a\xe2\x0be\xb1F\xb3{Kj\xed\t\xb9\xb3\xf2\x15\xf9\xd3E<7\xe1\xfb\x8d\x88\xc7\x9c\x81#l\x19\x07\xa5\x05\xdd\xdb\xf8\xc3\x1c\xff\xa7\x94\x9c\xc8]t~\x8e$/q\x10`g\xc1\xa1Q\xea|s\xd7\xaa\x9f\x0f\x9c#Og\x13\x95\xda}\x00\xc4\xf1\xc0S\xb8q\x9a\xddT\xffUY~\x07\x81\xc3]Qc\xe6\x7f\r\xd3bd\x0cG. e\xb2\xae\x80>,C\x81\xad\x83\xe4\xb5\x1f\x8c\xec\x8a$\xe3\x0c\xf2r\xa9\xed\xaf[[\xddL\xdd\x16\xf7\x1d&\x7f\xd8C\x0f\xc72\xcb\xd3J\xb4\x8a\xb9\n(\x86\x95B\xf4\xcd\x1b\xd5\xe4\xab\xb2G\xf1\xce!\x07@\xf0-\x1d\xbf\xcbD\xd3\x89\x07\\%\xe7\xce\x1d\x16z\xa4a\xf2\xedS\xcej\xd9uC]\x84\xa2\x801\xce\xf71\x14-\xc1\r}\xc8\xc6\x03W\xd4$\x9e\xe8QO\xec\xb3\x9e\x83\xa6\x90tu\x15X]\xb8!\x7f\xcaQ&bu\xb0!^\xcd\x99P\xbe\xe46J]\xa1\xa3\x8c\xcd\xc8\xca\xf8\xec\n\xc8\x8c\xc0\xd2\xd2$8\x95\x9d\x81\xdb\x9e\x05\x8f\x84\xaa\xfa\x07ci\xd5\xcb4\x1c\x9b\t\xe6\x02`M\xe1\x8e%P+.BI8\xcd\x94WT\x14~B5\x8fy\xafu\xdd\x10\x8dj"_\x9f\xd8\x84\xc3\xb9\xcb\x19m\x7f\xef`\x11\x06B\xd6g\xea\xe1\xc2\xc3\x19@\x0f\xd59\xf0\x01D\xfe\x08\x9ep~pE\xb1\x81\xd4\x7f\x9f>?_\xb2C\xfc%\x98\xfe,c!\x9c\x96\xaeH\xcc\x1cic(\xe6\x80)\x90W\x07\xfaOmQ:v\x87H[\xaa?\xf4Y\x89\xbaz\x96B\x96\xfbZ\x98\xf9Q\xfe\xe8\x1a\xf8.n\xca*\xc3`\xaa\x91}\xa4\xe7\x87\xf7\xff\xf4^\xf8\x01:r\xc8(\xa8\xfdf|WY[\xa7\xf4w\xfa@\xbc\\^w\xa8\xeb\x82\xe35G\xf7\x06f\xe34\x9aA\xbc\x1e\x88\xa84\xd2]\x95\x9d\xa6%T\xbb\x0bt\x93I\xc0+7\xa31\xa3?9\xb0B>\xd6J\ttT\x8f\xf8\x19\xec`\xab\xc0\xab\xab\x93G;UN)\x1b|\x0b\xde\x1d{\xffPLz\x08\xe4-f\\\xf1\xafM\xb7\x9e$:\xd8\xd2;\xfe\xc1\x02c\xa3\x16\x9c\xc5\x9b\xcdt\xb7\x99!\x02\xddL\xd62\xb9\xe7\xb4\x181\xe1S\xedJ\x83\x0b[\xc0B@\x93\x15\xcf\xb0\xaaS\xdd\xbfp{\x1cW0\xd2\x9d\x8aO\xaa\n\xba\x0e\xecm\x92\xdd\x84\x84\nlN\xd8\xe2n\xc7^\x85\x9c\xca\x0e\x05K\x8f\xb41\x9d\x07A\x812\xd4\xb4\'\x16:\xedG\x9d\xbfY\xd3\x05\xef\xd5hq\xaf\x88\xde\x068)\xca\x8b\xd4\x10pE\xc8e\x04i<\x18^\x1b\x7f\xfdJaP\x1f\xd4R\x10\xc5\xae\xe6Q\x08\x8b\xbcy*\x15\xb8\xb7G\xc1R\\\t\xbb\x93g\x9f\x1e\xdb7j\x02h\xe5\\@\xf0A\xfc^\x84U \xc9\xa7\x92%W\x0ep\x03\xb95gH!N\xfcjG\xb4\xc72\x0eq\x10W\xdd\xe8^Y\xa2p\xab\xd1\xcf\xe4\xd8q\xa9\xce]\xf9\xdbJ\x8e\xe9\xaaS\xed6y\x93~2\x8e\xbbx\x0eIF5\x16\xdco\xd1\xec;\tqm \xd3\x80\x93\xcc\xb38\xe9\x12\x1a\xb1\xbd\x1f\xb5\x80\xef\x8bR@f\'QF\xac\xd1\x1e\x18\x1e-\x80\x85\x82\xe8\x9f\xb6|E\xc1c\xe4\xcc\xd0\x96\x88\xc5\xa0\x83"\xa3\xa69\xfb\x10\xc2XN\x1e9\xe9\xeeB\x01\x00\xe8w\xc3\xb1\xb6\x9e\xbd\x8d\n\x92\x94Vc\x85mjA=\x92?\x12\x17\xa2nJV$\xcd\xea/XO\x00\xf9\x8b\xdc\xf0\xcef\xc8\x06\x94*M\xce\xfb\x0b\xe5/SC\xad\xed\xdfu\x95\x83S\xdc\xa9\xad:\xdc\t\xc8r\x9c\x1f\x16\x85\x0e\x11\xff\xa4\xeeJI\xddK\xf4!e\x9c\xb61\x81\xe90\x13\x06\t*\x86H\x86\xf7\r\x01\t\x151\x06\x04\x04\x01\x00\x00\x000W\x06\t*\x86H\x86\xf7\r\x01\t\x141J\x1eH\x00b\x00d\x008\x003\x007\x009\x00c\x001\x00-\x00a\x006\x005\x00c\x00-\x004\x003\x00e\x00e\x00-\x008\x001\x000\x00a\x00-\x005\x00a\x002\x004\x003\x004\x006\x003\x00f\x00c\x00e\x00c0y\x06\t+\x06\x01\x04\x01\x827\x11\x011l\x1ej\x00M\x00i\x00c\x00r\x00o\x00s\x00o\x00f\x00t\x00 \x00E\x00n\x00h\x00a\x00n\x00c\x00e\x00d\x00 \x00R\x00S\x00A\x00 \x00a\x00n\x00d\x00 \x00A\x00E\x00S\x00 \x00C\x00r\x00y\x00p\x00t\x00o\x00g\x00r\x00a\x00p\x00h\x00i\x00c\x00 \x00P\x00r\x00o\x00v\x00i\x00d\x00e\x00r0\x82\x03\xbf\x06\t*\x86H\x86\xf7\r\x01\x07\x06\xa0\x82\x03\xb00\x82\x03\xac\x02\x01\x000\x82\x03\xa5\x06\t*\x86H\x86\xf7\r\x01\x07\x010\x1c\x06\n*\x86H\x86\xf7\r\x01\x0c\x01\x060\x0e\x04\x08\xd0~p\xe13w{_\x02\x02\x07\xd0\x80\x82\x03x\x1e]uF\xa8\xb8\xdf\xf4\xb0x4\xf4\xe3r\xea\xee\xc7\x83k;\x07.\xde!W\x98b\x0f]\x87v\x06^\xe2\xc4Z\xd9\x16\xee\xe4z\xb5\xa5^R\xf8!V\x01\xcb\x90\xbf\x05\xdfWN\x8c\xe2\x059\xba\x02\xa0\xf3\xb0d\x0f\x94 |\xe89\x07\xa8\xff\x86\xc5\x12\x8c\xb0Pa\xaa\xb2\xb3\xbe\xe7\xd5 \xc7\xf6\xa5\xa0\xe9\x87\x93\x16\xea\xebPV\xc9\xda\x97B\x1e\xa1M)\x1c^\xe7\xfb\xbe\x00q;\x8e\xf4\r\xf3q\x87\xb9)g\xc8\xa1\x07i\x06\xca\xc8\x97S\xd3\xd3ihO\xaf(\xd7\xa0\t\xefo\x10\xb8\xeaE\xfc\xdf\x18\xf0\xd4\xe5\xbc\xbc\x12\x11\xc2p\xb4\xfd\xf8\x1b\x0e\xa8\xa9.N[?\xd2\xf3\x0f\xda\x1cp\x99\x86\xbe\xc3\xa8t\\\xa4\x88?\x90\xb3\xa3qvh\xa5\x06\xf1a\xb6%\x18\x8eqc2\x04\xc7\xc0>i\xdc\x8fI\xbc\xb0\x82\xdc\x87j\xd9\xff\x05e{\xd2\x01\x0e\xf0\x91\x7f\xcb\xf8\xa27\xfb\xd1\x9c\x96\xadS\x97\x17\xb0\xf3F\xb4\xc6\xc3(/\xc4\x88\xdf\xf3M@n\xc4\xd88\xd0%hq\x0c\xae\xab\xbf\x91\xe6\xff\xd4np\xed\x9a\xab\x04\x83^/!\xf7\xf5\x91\xa54\xc9n\xb14\xb6\x8d\xdbQB-\x04\xf8\x98H\x98\xe10Ksw\xac\x0cc@\x92\x10\x1cA\x96\xb39gljl6\xd4cz\xd8it\x1e\xedZ]\xfa\x0c\x93\xce\x97\xcd"\x93e\xf8\xb1\x02b\xb9 \xb5\xc5W\x12\xf5\xe7\x81s\xf9$+\xf4\xee\x13\xdeQ\x02Cq\x15x\xe3\xd8\xbe&\xa4,k0U\xa9\x87\x15\xc7\x86\x82\xb0@Q\xfb\xbb\xf6\xa11\xbf\x1d[,\xec\x16`\xc5\xc2\x0f/\n\x8f\x9f\xebf\xfa\xf3\xcfW\x9d\xedD\xf4\xdf\x8c\x88\xc3{\x89s\xf6\xb20+\xa2J\xffx\xbd\xad\x9a9/\xee\xc2x?\xce\xc2{\xb8\x1b)\x0e\x8ao\xe3\x86{%n.~\x8c\x95\xc9\x85[\xfa\x1a070\x1f0\x07\x06\x05+\x0e\x03\x02\x1a\x04\x14\xc2\xeb\x94\x0bl\xc6q36\x8c\x8b\xfb\x87\xa2N\x0e\xa4\x00\x11.\x04\x14R)\xd7xhz=l\x84,\xcc\x07\xa6\xef\xd7i\xb0\xe9\x13o' async def _import_common_certificate(self, client, cert_name): cert_password = "123" cert_policy = CertificatePolicy( issuer_name="Self", subject="CN=DefaultPolicy", exportable=True, key_type="RSA", key_size=2048, reuse_key=False, content_type="application/x-pkcs12", validity_in_months=12, key_usage=["digitalSignature", "keyEncipherment"], ) return await client.import_certificate( cert_name, CertificateClientTests.CERT_CONTENT_PASSWORD_ENODED, policy=cert_policy, password=cert_password ) def _validate_certificate_operation(self, pending_cert_operation, vault, cert_name, original_cert_policy): self.assertIsNotNone(pending_cert_operation) self.assertIsNotNone(pending_cert_operation.csr) self.assertEqual(original_cert_policy.issuer_name, pending_cert_operation.issuer_name) pending_id = parse_key_vault_certificate_id(pending_cert_operation.id) self.assertEqual(pending_id.vault_url.strip("/"), vault.strip("/")) self.assertEqual(pending_id.name, cert_name) def _validate_certificate_bundle(self, cert, cert_name, cert_policy): self.assertIsNotNone(cert) self.assertEqual(cert_name, cert.name) self.assertIsNotNone(cert.cer) self.assertIsNotNone(cert.policy) self._validate_certificate_policy(cert_policy, cert_policy) def _validate_certificate_policy(self, a, b): self.assertEqual(a.issuer_name, b.issuer_name) self.assertEqual(a.subject, b.subject) self.assertEqual(a.exportable, b.exportable) self.assertEqual(a.key_type, b.key_type) self.assertEqual(a.key_size, b.key_size) self.assertEqual(a.reuse_key, b.reuse_key) self.assertEqual(a.key_curve_name, b.key_curve_name) if a.enhanced_key_usage: self.assertEqual(set(a.enhanced_key_usage), set(b.enhanced_key_usage)) if a.key_usage: self.assertEqual(set(a.key_usage), set(b.key_usage)) self.assertEqual(a.content_type, b.content_type) self.assertEqual(a.validity_in_months, b.validity_in_months) self.assertEqual(a.certificate_type, b.certificate_type) self.assertEqual(a.certificate_transparency, b.certificate_transparency) self._validate_sans(a, b) if a.lifetime_actions: self._validate_lifetime_actions(a.lifetime_actions, b.lifetime_actions) def _validate_sans(self, a, b): if a.san_dns_names: self.assertEqual(set(a.san_dns_names), set(b.san_dns_names)) if a.san_emails: self.assertEqual(set(a.san_emails), set(b.san_emails)) if a.san_user_principal_names: self.assertEqual(set(a.san_user_principal_names), set(b.san_user_principal_names)) def _validate_lifetime_actions(self, a, b): self.assertEqual(len(a), len(b)) for a_entry in a: b_entry = next(x for x in b if x.action == a_entry.action) self.assertEqual(a_entry.lifetime_percentage, b_entry.lifetime_percentage) self.assertEqual(a_entry.days_before_expiry, b_entry.days_before_expiry) async def _validate_certificate_list(self, a, b): async for cert in b: if cert.id in a.keys(): del a[cert.id] else: assert False, "Returned certificate with id {} not found in list of original certificates".format( cert.id ) self.assertEqual(len(a), 0) def _validate_certificate_contacts(self, a, b): self.assertEqual(len(a), len(b)) for a_entry in a: b_entry = next(x for x in b if x.email == a_entry.email) self.assertEqual(a_entry.name, b_entry.name) self.assertEqual(a_entry.phone, b_entry.phone) def _admin_contact_equal(self, a, b): return a.first_name == b.first_name and a.last_name == b.last_name and a.email == b.email and a.phone == b.phone def _validate_certificate_issuer(self, a, b): self.assertEqual(a.provider, b.provider) self.assertEqual(a.account_id, b.account_id) self.assertEqual(len(a.admin_contacts), len(b.admin_contacts)) for a_admin_contact in a.admin_contacts: b_admin_contact = next( (ad for ad in b.admin_contacts if self._admin_contact_equal(a_admin_contact, ad)), None ) self.assertIsNotNone(b_admin_contact) self.assertEqual(a.password, b.password) self.assertEqual(a.organization_id, b.organization_id) def _validate_certificate_issuer_properties(self, a, b): self.assertEqual(a.id, b.id) self.assertEqual(a.name, b.name) self.assertEqual(a.provider, b.provider) @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer() async def test_crud_operations(self, client, **kwargs): cert_name = self.get_resource_name("cert") lifetime_actions = [LifetimeAction(lifetime_percentage=80, action=CertificatePolicyAction.auto_renew)] cert_policy = CertificatePolicy( issuer_name="Self", subject="CN=DefaultPolicy", exportable=True, key_type=KeyType.rsa, key_size=2048, reuse_key=False, content_type=CertificateContentType.pkcs12, lifetime_actions=lifetime_actions, validity_in_months=12, key_usage=[KeyUsageType.digital_signature, KeyUsageType.key_encipherment], ) # create certificate cert = await client.create_certificate(certificate_name=cert_name, policy=CertificatePolicy.get_default()) self._validate_certificate_bundle(cert=cert, cert_name=cert_name, cert_policy=cert_policy) self.assertEqual( (await client.get_certificate_operation(certificate_name=cert_name)).status.lower(), "completed" ) # get certificate cert = await client.get_certificate(certificate_name=cert_name) self._validate_certificate_bundle(cert=cert, cert_name=cert_name, cert_policy=cert_policy) # update certificate, ensuring the new updated_on value is at least one second later than the original if self.is_live: await asyncio.sleep(1) tags = {"tag1": "updated_value1"} updated_cert = await client.update_certificate_properties(cert_name, tags=tags) self._validate_certificate_bundle(cert=updated_cert, cert_name=cert_name, cert_policy=cert_policy) self.assertEqual(tags, updated_cert.properties.tags) self.assertEqual(cert.id, updated_cert.id) self.assertNotEqual(cert.properties.updated_on, updated_cert.properties.updated_on) # delete certificate deleted_cert_bundle = await client.delete_certificate(certificate_name=cert_name) self._validate_certificate_bundle(cert=deleted_cert_bundle, cert_name=cert_name, cert_policy=cert_policy) # get certificate returns not found try: await client.get_certificate_version(cert_name, deleted_cert_bundle.properties.version) self.fail("Get should fail") except Exception as ex: if not hasattr(ex, "message") or "not found" not in ex.message.lower(): raise ex @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer() async def test_import_certificate_not_password_encoded_no_policy(self, client): # If a certificate is not password encoded, we can import the certificate # without passing in 'password' certificate = await client.import_certificate( certificate_name="importNotPasswordEncodedCertificate", certificate_bytes=CertificateClientTests.CERT_CONTENT_NOT_PASSWORD_ENCODED, ) self.assertIsNotNone(certificate.policy) @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer() async def test_import_certificate_password_encoded_no_policy(self, client): # If a certificate is password encoded, we have to pass in 'password' # when importing the certificate certificate = await client.import_certificate( certificate_name="importPasswordEncodedCertificate", certificate_bytes=CertificateClientTests.CERT_CONTENT_PASSWORD_ENODED, password="123", ) self.assertIsNotNone(certificate.policy) @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer() async def test_list(self, client, **kwargs): max_certificates = self.list_test_size expected = {} # import some certificates for x in range(max_certificates): cert_name = self.get_resource_name("cert{}".format(x)) error_count = 0 try: cert_bundle = await self._import_common_certificate(client=client, cert_name=cert_name) # Going to remove the ID from the last '/' onwards. This is because list_properties_of_certificates # doesn't return the version in the ID cid = "/".join(cert_bundle.id.split("/")[:-1]) expected[cid] = cert_bundle except Exception as ex: if hasattr(ex, "message") and "Throttled" in ex.message: error_count += 1 await asyncio.sleep(2.5 * error_count) continue else: raise ex # list certificates returned_certificates = client.list_properties_of_certificates(max_page_size=max_certificates - 1) await self._validate_certificate_list(expected, returned_certificates) @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer() async def test_list_certificate_versions(self, client, **kwargs): cert_name = self.get_resource_name("certver") max_certificates = self.list_test_size expected = {} # import same certificates as different versions for x in range(max_certificates): error_count = 0 try: cert_bundle = await self._import_common_certificate(client=client, cert_name=cert_name) expected[cert_bundle.id.strip("/")] = cert_bundle except Exception as ex: if hasattr(ex, "message") and "Throttled" in ex.message: error_count += 1 await asyncio.sleep(2.5 * error_count) continue else: raise ex # list certificate versions await self._validate_certificate_list( expected, ( client.list_properties_of_certificate_versions( certificate_name=cert_name, max_page_size=max_certificates - 1 ) ), ) @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer() async def test_crud_contacts(self, client, **kwargs): contact_list = [ CertificateContact(email="admin@contoso.com", name="John Doe", phone="1111111111"), CertificateContact(email="admin2@contoso.com", name="John Doe2", phone="2222222222"), ] # create certificate contacts contacts = await client.set_contacts(contacts=contact_list) self._validate_certificate_contacts(contact_list, contacts) # get certificate contacts contacts = await client.get_contacts() self._validate_certificate_contacts(contact_list, contacts) # delete certificate contacts contacts = await client.delete_contacts() self._validate_certificate_contacts(contact_list, contacts) # get certificate contacts returns not found try: await client.get_contacts() self.fail("Get should fail") except Exception as ex: if not hasattr(ex, "message") or "not found" not in ex.message.lower(): raise ex @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer() async def test_recover_and_purge(self, client, **kwargs): certs = {} # create certificates to recover for i in range(self.list_test_size): cert_name = self.get_resource_name("certrec{}".format(str(i))) certs[cert_name] = await self._import_common_certificate(client=client, cert_name=cert_name) # create certificates to purge for i in range(self.list_test_size): cert_name = self.get_resource_name("certprg{}".format(str(i))) certs[cert_name] = await self._import_common_certificate(client=client, cert_name=cert_name) # delete all certificates for cert_name in certs.keys(): await client.delete_certificate(certificate_name=cert_name) # validate all our deleted certificates are returned by list_deleted_certificates deleted_certificates = client.list_deleted_certificates() deleted = [] async for c in deleted_certificates: deleted.append(parse_key_vault_certificate_id(source_id=c.id).name) self.assertTrue(all(c in deleted for c in certs.keys())) # recover select certificates for certificate_name in [c for c in certs.keys() if c.startswith("certrec")]: await client.recover_deleted_certificate(certificate_name=certificate_name) # purge select certificates for certificate_name in [c for c in certs.keys() if c.startswith("certprg")]: await client.purge_deleted_certificate(certificate_name) if not self.is_playback(): await asyncio.sleep(50) # validate none of our deleted certificates are returned by list_deleted_certificates deleted_certificates = client.list_deleted_certificates() deleted = [] async for c in deleted_certificates: deleted.append(parse_key_vault_certificate_id(source_id=c.id).name) self.assertTrue(not any(c in deleted for c in certs.keys())) # validate the recovered certificates expected = {k: v for k, v in certs.items() if k.startswith("certrec")} actual = {} for k in expected.keys(): actual[k] = await client.get_certificate_version(certificate_name=k, version="") self.assertEqual(len(set(expected.keys()) & set(actual.keys())), len(expected)) @pytest.mark.skip("Skipping because service doesn't allow cancellation of certificates with issuer 'Unknown'") @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer() async def test_async_request_cancellation_and_deletion(self, client, **kwargs): cert_name = "asyncCanceledDeletedCert" cert_policy = CertificatePolicy.get_default() # create certificate await client.create_certificate(certificate_name=cert_name, policy=cert_policy) # cancel certificate operation cancel_operation = await client.cancel_certificate_operation(cert_name) self.assertTrue(hasattr(cancel_operation, "cancellation_requested")) self.assertTrue(cancel_operation.cancellation_requested) self._validate_certificate_operation( pending_cert_operation=cancel_operation, vault=client.vault_url, cert_name=cert_name, original_cert_policy=cert_policy, ) cancelled = False for _ in range(10): if (await client.get_certificate_operation(cert_name)).status.lower() == "cancelled": cancelled = True break await asyncio.sleep(10) self.assertTrue(cancelled) retrieved_operation = await client.get_certificate_operation(certificate_name=cert_name) self.assertTrue(hasattr(retrieved_operation, "cancellation_requested")) self.assertTrue(retrieved_operation.cancellation_requested) self._validate_certificate_operation( pending_cert_operation=retrieved_operation, vault=client.vault_url, cert_name=cert_name, original_cert_policy=cert_policy, ) # delete certificate operation deleted_operation = await client.delete_certificate_operation(cert_name) self.assertIsNotNone(deleted_operation) self._validate_certificate_operation( pending_cert_operation=deleted_operation, vault=client.vault_url, cert_name=cert_name, original_cert_policy=cert_policy, ) try: await client.get_certificate_operation(certificate_name=cert_name) self.fail("Get should fail") except Exception as ex: if not hasattr(ex, "message") or "not found" not in ex.message.lower(): raise ex # delete cancelled certificate await client.delete_certificate(cert_name) @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer() async def test_policy(self, client, **kwargs): cert_name = "policyCertificate" cert_policy = CertificatePolicy( issuer_name="Self", subject="CN=DefaultPolicy", exportable=True, key_type=KeyType.rsa, key_size=2048, reuse_key=True, enhanced_key_usage=["1.3.6.1.5.5.7.3.1", "1.3.6.1.5.5.7.3.2"], key_usage=[KeyUsageType.decipher_only], content_type=CertificateContentType.pkcs12, validity_in_months=12, lifetime_actions=[LifetimeAction(action=CertificatePolicyAction.email_contacts, lifetime_percentage=98)], certificate_transparency=False, san_dns_names=["sdk.azure-int.net"], ) # get certificate policy await client.create_certificate(cert_name, cert_policy) returned_policy = await client.get_certificate_policy(certificate_name=cert_name) self._validate_certificate_policy(cert_policy, returned_policy) cert_policy._key_type = KeyType.ec cert_policy._key_size = 256 cert_policy._key_curve_name = KeyCurveName.p_256 returned_policy = await client.update_certificate_policy(certificate_name=cert_name, policy=cert_policy) self._validate_certificate_policy(cert_policy, returned_policy) @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer() async def test_get_pending_certificate_signing_request(self, client, **kwargs): cert_name = "unknownIssuerCert" # get pending certificate signing request await client.create_certificate(certificate_name=cert_name, policy=CertificatePolicy.get_default()) operation = await client.get_certificate_operation(certificate_name=cert_name) pending_version_csr = operation.csr self.assertEqual((await client.get_certificate_operation(certificate_name=cert_name)).csr, pending_version_csr) @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer() async def test_backup_restore(self, client, **kwargs): cert_name = self.get_resource_name("cert") policy = CertificatePolicy.get_default() policy._san_user_principal_names = ["john.doe@domain.com"] # create certificate await client.create_certificate(certificate_name=cert_name, policy=policy) # create a backup certificate_backup = await client.backup_certificate(certificate_name=cert_name) # delete the certificate await client.delete_certificate(certificate_name=cert_name) # purge the certificate await client.purge_deleted_certificate(certificate_name=cert_name) # restore certificate restore_function = functools.partial(client.restore_certificate_backup, certificate_backup) restored_certificate = await self._poll_until_no_exception( restore_function, expected_exception=ResourceExistsError ) self._validate_certificate_bundle(cert=restored_certificate, cert_name=cert_name, cert_policy=policy) @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer() async def test_crud_issuer(self, client, **kwargs): issuer_name = "issuer" admin_contacts = [ AdministratorContact(first_name="John", last_name="Doe", email="admin@microsoft.com", phone="4255555555") ] # create certificate issuer issuer = await client.create_issuer( issuer_name, "Test", account_id="keyvaultuser", admin_contacts=admin_contacts, enabled=True ) expected = CertificateIssuer( provider="Test", account_id="keyvaultuser", admin_contacts=admin_contacts, issuer_id=client.vault_url + "/certificates/issuers/" + issuer_name, ) self._validate_certificate_issuer(expected, issuer) # get certificate issuer issuer = await client.get_issuer(issuer_name=issuer_name) self._validate_certificate_issuer(expected, issuer) # list certificate issuers await client.create_issuer( issuer_name=issuer_name + "2", provider="Test", account_id="keyvaultuser2", admin_contacts=admin_contacts, enabled=True, ) expected_base_1 = IssuerProperties( issuer_id=client.vault_url + "/certificates/issuers/" + issuer_name, provider="Test" ) expected_base_2 = IssuerProperties( issuer_id=client.vault_url + "/certificates/issuers/" + issuer_name + "2", provider="Test" ) expected_issuers = [expected_base_1, expected_base_2] issuers = client.list_properties_of_issuers() async for issuer in issuers: exp_issuer = next((i for i in expected_issuers if i.name == issuer.name), None) self.assertIsNotNone(exp_issuer) self._validate_certificate_issuer_properties(exp_issuer, issuer) expected_issuers.remove(exp_issuer) self.assertEqual(len(expected_issuers), 0) # update certificate issuer admin_contacts = [ AdministratorContact(first_name="Jane", last_name="Doe", email="admin@microsoft.com", phone="4255555555") ] expected = CertificateIssuer( provider="Test", account_id="keyvaultuser", admin_contacts=admin_contacts, issuer_id=client.vault_url + "/certificates/issuers/" + issuer_name, ) issuer = await client.update_issuer(issuer_name, admin_contacts=admin_contacts) self._validate_certificate_issuer(expected, issuer) # delete certificate issuer await client.delete_issuer(issuer_name=issuer_name) # get certificate issuer returns not found try: await client.get_issuer(issuer_name=issuer_name) self.fail("Get should fail") except Exception as ex: if not hasattr(ex, "message") or "not found" not in ex.message.lower(): raise ex @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer(client_kwargs={"logging_enable": True}) async def test_logging_enabled(self, client, **kwargs): mock_handler = MockHandler() logger = logging.getLogger("azure") logger.addHandler(mock_handler) logger.setLevel(logging.DEBUG) await client.create_issuer(issuer_name="cert-name", provider="Test") for message in mock_handler.messages: if message.levelname == "DEBUG" and message.funcName == "on_request": try: body = json.loads(message.message) if body["provider"] == "Test": return except (ValueError, KeyError): # this means the message is not JSON or has no kty property pass assert False, "Expected request body wasn't logged" @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer() async def test_logging_disabled(self, client, **kwargs): mock_handler = MockHandler() logger = logging.getLogger("azure") logger.addHandler(mock_handler) logger.setLevel(logging.DEBUG) await client.create_issuer(issuer_name="cert-name", provider="Test") for message in mock_handler.messages: if message.levelname == "DEBUG" and message.funcName == "on_request": try: body = json.loads(message.message) assert body["provider"] != "Test", "Client request body was logged" except (ValueError, KeyError): # this means the message is not JSON or has no kty property pass @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer() async def test_get_certificate_version(self, client, **kwargs): cert_name = self.get_resource_name("cert") policy = CertificatePolicy.get_default() for _ in range(self.list_test_size): await client.create_certificate(cert_name, policy) async for version_properties in client.list_properties_of_certificate_versions(cert_name): cert = await client.get_certificate_version(version_properties.name, version_properties.version) # This isn't factored out into a helper method because the properties are not exactly equal. # get_certificate_version sets "recovery_days" and "recovery_level" but the list method does not. # (This is Key Vault's behavior, not an SDK limitation.) assert version_properties.created_on == cert.properties.created_on assert version_properties.enabled == cert.properties.enabled assert version_properties.expires_on == cert.properties.expires_on assert version_properties.id == cert.properties.id assert version_properties.name == cert.properties.name assert version_properties.not_before == cert.properties.not_before assert version_properties.tags == cert.properties.tags assert version_properties.updated_on == cert.properties.updated_on assert version_properties.vault_url == cert.properties.vault_url assert version_properties.version == cert.properties.version assert version_properties.x509_thumbprint == cert.properties.x509_thumbprint @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer(client_kwargs={"api_version": ApiVersion.V2016_10_01}) async def test_list_properties_of_certificates_2016_10_01(self, client, **kwargs): certs = client.list_properties_of_certificates() async for cert in certs: pass with pytest.raises(NotImplementedError) as excinfo: certs = client.list_properties_of_certificates(include_pending=True) async for cert in certs: pass assert "The 'include_pending' parameter to `list_properties_of_certificates` is only available for API versions v7.0 and up" in str(excinfo.value) @ResourceGroupPreparer(random_name_enabled=True) @KeyVaultPreparer() @KeyVaultClientPreparer(client_kwargs={"api_version": ApiVersion.V2016_10_01}) async def test_list_deleted_certificates_2016_10_01(self, client, **kwargs): certs = client.list_deleted_certificates() async for cert in certs: pass with pytest.raises(NotImplementedError) as excinfo: certs = client.list_deleted_certificates(include_pending=True) async for cert in certs: pass assert "The 'include_pending' parameter to `list_deleted_certificates` is only available for API versions v7.0 and up" in str(excinfo.value) def test_service_headers_allowed_in_logs(): service_headers = {"x-ms-keyvault-network-info", "x-ms-keyvault-region", "x-ms-keyvault-service-version"} client = CertificateClient("...", object()) assert service_headers.issubset(client._client._config.http_logging_policy.allowed_header_names) def test_custom_hook_policy(): class CustomHookPolicy(object): pass client = CertificateClient("...", object(), custom_hook_policy=CustomHookPolicy()) assert isinstance(client._client._config.custom_hook_policy, CustomHookPolicy)