本文整理汇总了Python中nova.crypto.fetch_ca函数的典型用法代码示例。如果您正苦于以下问题：Python fetch_ca函数的具体用法？Python fetch_ca怎么用？Python fetch_ca使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了fetch_ca函数的17个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_can_generate_x509
def test_can_generate_x509(self):
# NOTE(todd): this doesn't assert against the auth manager
# so it probably belongs in crypto_unittest
# but I'm leaving it where I found it.
with user_and_project_generator(self.manager) as (user, project):
# NOTE(vish): Setup runs genroot.sh if it hasn't been run
cloud.CloudController().setup()
_key, cert_str = crypto.generate_x509_cert(user.id, project.id)
LOG.debug(cert_str)
full_chain = crypto.fetch_ca(project_id=project.id, chain=True)
int_cert = crypto.fetch_ca(project_id=project.id, chain=False)
cloud_cert = crypto.fetch_ca()
LOG.debug("CA chain:\n\n =====\n%s\n\n=====", full_chain)
signed_cert = X509.load_cert_string(cert_str)
chain_cert = X509.load_cert_string(full_chain)
int_cert = X509.load_cert_string(int_cert)
cloud_cert = X509.load_cert_string(cloud_cert)
self.assertTrue(signed_cert.verify(chain_cert.get_pubkey()))
self.assertTrue(signed_cert.verify(int_cert.get_pubkey()))
if not FLAGS.use_project_ca:
self.assertTrue(signed_cert.verify(cloud_cert.get_pubkey()))
else:
self.assertFalse(signed_cert.verify(cloud_cert.get_pubkey()))
开发者ID:lovocas，项目名称:nova，代码行数:25，代码来源:test_auth.py
示例2: test_encrypt_decrypt_x509
def test_encrypt_decrypt_x509(self):
with utils.tempdir() as tmpdir:
self.flags(ca_path=tmpdir)
project_id = "fake"
crypto.ensure_ca_filesystem()
cert = crypto.fetch_ca(project_id)
public_key = os.path.join(tmpdir, "public.pem")
with open(public_key, 'w') as keyfile:
keyfile.write(cert)
text = "some @#!%^* test text"
process_input = text.encode("ascii") if six.PY3 else text
enc, _err = utils.execute('openssl',
'rsautl',
'-certin',
'-encrypt',
'-inkey', '%s' % public_key,
process_input=process_input,
binary=True)
dec = crypto.decrypt_text(project_id, enc)
self.assertIsInstance(dec, bytes)
if six.PY3:
dec = dec.decode('ascii')
self.assertEqual(text, dec)
开发者ID:B3n0n3，项目名称:nova，代码行数:26，代码来源:test_crypto.py
示例3: get_credentials
def get_credentials(self):
rc = self.generate_rc()
private_key, signed_cert = self.generate_x509_cert()
configfile = open(FLAGS.vpn_client_template,"r")
s = Template(configfile.read())
configfile.close()
config = s.substitute(keyfile=FLAGS.credential_key_file,
certfile=FLAGS.credential_cert_file,
ip=self.vpn_ip,
port=self.vpn_port)
tmpdir = tempfile.mkdtemp()
zf = os.path.join(tmpdir, "temp.zip")
zippy = zipfile.ZipFile(zf, 'w')
zippy.writestr(FLAGS.credential_rc_file, rc)
zippy.writestr(FLAGS.credential_key_file, private_key)
zippy.writestr(FLAGS.credential_cert_file, signed_cert)
zippy.writestr("nebula-client.conf", config)
zippy.writestr(FLAGS.ca_file, crypto.fetch_ca(self.id))
zippy.close()
with open(zf, 'rb') as f:
buffer = f.read()
shutil.rmtree(tmpdir)
return buffer
开发者ID:joshuamckenty，项目名称:pinet，代码行数:26，代码来源:users.py
示例4: test_011_can_generate_x509
def test_011_can_generate_x509(self):
# MUST HAVE RUN CLOUD SETUP BY NOW
self.cloud = cloud.CloudController()
self.cloud.setup()
private_key, signed_cert_string = self.users.get_user('test1').generate_x509_cert()
logging.debug(signed_cert_string)
# Need to verify that it's signed by the right intermediate CA
full_chain = crypto.fetch_ca(username='test1', chain=True)
int_cert = crypto.fetch_ca(username='test1', chain=False)
cloud_cert = crypto.fetch_ca()
logging.debug("CA chain:\n\n =====\n%s\n\n=====" % full_chain)
signed_cert = X509.load_cert_string(signed_cert_string)
chain_cert = X509.load_cert_string(full_chain)
int_cert = X509.load_cert_string(int_cert)
cloud_cert = X509.load_cert_string(cloud_cert)
self.assertTrue(signed_cert.verify(chain_cert.get_pubkey()))
self.assertTrue(signed_cert.verify(int_cert.get_pubkey()))
self.assertFalse(signed_cert.verify(cloud_cert.get_pubkey()))
开发者ID:joshuamckenty，项目名称:pinet，代码行数:19，代码来源:users_unittest.py
示例5: test_can_generate_x509
def test_can_generate_x509(self):
tmpdir = tempfile.mkdtemp()
self.flags(ca_path=tmpdir)
try:
crypto.ensure_ca_filesystem()
_key, cert_str = crypto.generate_x509_cert('fake', 'fake')
project_cert = crypto.fetch_ca(project_id='fake')
cloud_cert = crypto.fetch_ca()
# TODO(vish): This will need to be replaced with something else
# when we remove M2Crypto
signed_cert = X509.load_cert_string(cert_str)
project_cert = X509.load_cert_string(project_cert)
cloud_cert = X509.load_cert_string(cloud_cert)
self.assertTrue(signed_cert.verify(project_cert.get_pubkey()))
if not FLAGS.use_project_ca:
self.assertTrue(signed_cert.verify(cloud_cert.get_pubkey()))
else:
self.assertFalse(signed_cert.verify(cloud_cert.get_pubkey()))
finally:
shutil.rmtree(tmpdir)
开发者ID:KarimAllah，项目名称:nova，代码行数:22，代码来源:test_crypto.py
示例6: test_209_can_generate_x509
def test_209_can_generate_x509(self):
# MUST HAVE RUN CLOUD SETUP BY NOW
self.cloud = cloud.CloudController()
self.cloud.setup()
_key, cert_str = self.manager._generate_x509_cert('test1', 'testproj')
logging.debug(cert_str)
# Need to verify that it's signed by the right intermediate CA
full_chain = crypto.fetch_ca(project_id='testproj', chain=True)
int_cert = crypto.fetch_ca(project_id='testproj', chain=False)
cloud_cert = crypto.fetch_ca()
logging.debug("CA chain:\n\n =====\n%s\n\n=====" % full_chain)
signed_cert = X509.load_cert_string(cert_str)
chain_cert = X509.load_cert_string(full_chain)
int_cert = X509.load_cert_string(int_cert)
cloud_cert = X509.load_cert_string(cloud_cert)
self.assertTrue(signed_cert.verify(chain_cert.get_pubkey()))
self.assertTrue(signed_cert.verify(int_cert.get_pubkey()))
if not FLAGS.use_intermediate_ca:
self.assertTrue(signed_cert.verify(cloud_cert.get_pubkey()))
else:
self.assertFalse(signed_cert.verify(cloud_cert.get_pubkey()))
开发者ID:ChristopherMacGown，项目名称:my-nova，代码行数:23，代码来源:auth_unittest.py
示例7: get_credentials
def get_credentials(self, user, project=None, use_dmz=True):
"""Get credential zip for user in project"""
if not isinstance(user, User):
user = self.get_user(user)
if project is None:
project = user.id
pid = Project.safe_id(project)
private_key, signed_cert = crypto.generate_x509_cert(user.id, pid)
tmpdir = tempfile.mkdtemp()
zf = os.path.join(tmpdir, "temp.zip")
zippy = zipfile.ZipFile(zf, 'w')
if use_dmz and FLAGS.region_list:
regions = {}
for item in FLAGS.region_list:
region, _sep, region_host = item.partition("=")
regions[region] = region_host
else:
regions = {'nova': FLAGS.ec2_host}
for region, host in regions.iteritems():
rc = self.__generate_rc(user,
pid,
use_dmz,
host)
zippy.writestr(FLAGS.credential_rc_file % region, rc)
zippy.writestr(FLAGS.credential_key_file, private_key)
zippy.writestr(FLAGS.credential_cert_file, signed_cert)
(vpn_ip, vpn_port) = self.get_project_vpn_data(project)
if vpn_ip:
configfile = open(FLAGS.vpn_client_template, "r")
s = string.Template(configfile.read())
configfile.close()
config = s.substitute(keyfile=FLAGS.credential_key_file,
certfile=FLAGS.credential_cert_file,
ip=vpn_ip,
port=vpn_port)
zippy.writestr(FLAGS.credential_vpn_file, config)
else:
LOG.warn(_("No vpn data for project %s"), pid)
zippy.writestr(FLAGS.ca_file, crypto.fetch_ca(pid))
zippy.close()
with open(zf, 'rb') as f:
read_buffer = f.read()
shutil.rmtree(tmpdir)
return read_buffer
开发者ID:chopachom，项目名称:nova，代码行数:49，代码来源:manager.py
示例8: test_encrypt_decrypt_x509
def test_encrypt_decrypt_x509(self):
with utils.tempdir() as tmpdir:
self.flags(ca_path=tmpdir)
project_id = "fake"
crypto.ensure_ca_filesystem()
cert = crypto.fetch_ca(project_id)
public_key = os.path.join(tmpdir, "public.pem")
with open(public_key, "w") as keyfile:
keyfile.write(cert)
text = "some @#!%^* test text"
enc, _err = utils.execute(
"openssl", "rsautl", "-certin", "-encrypt", "-inkey", "%s" % public_key, process_input=text
)
dec = crypto.decrypt_text(project_id, enc)
self.assertEqual(text, dec)
开发者ID:nitishb，项目名称:nova，代码行数:15，代码来源:test_crypto.py
示例9: test_can_generate_x509
def test_can_generate_x509(self):
with utils.tempdir() as tmpdir:
self.flags(ca_path=tmpdir)
crypto.ensure_ca_filesystem()
_key, cert_str = crypto.generate_x509_cert("fake", "fake")
project_cert = crypto.fetch_ca(project_id="fake")
signed_cert_file = os.path.join(tmpdir, "signed")
with open(signed_cert_file, "w") as keyfile:
keyfile.write(cert_str)
project_cert_file = os.path.join(tmpdir, "project")
with open(project_cert_file, "w") as keyfile:
keyfile.write(project_cert)
enc, err = utils.execute("openssl", "verify", "-CAfile", project_cert_file, "-verbose", signed_cert_file)
self.assertFalse(err)
开发者ID:nitishb，项目名称:nova，代码行数:18，代码来源:test_crypto.py
示例10: test_can_generate_x509
def test_can_generate_x509(self):
with utils.tempdir() as tmpdir:
self.flags(ca_path=tmpdir)
crypto.ensure_ca_filesystem()
_key, cert_str = crypto.generate_x509_cert('fake', 'fake')
project_cert = crypto.fetch_ca(project_id='fake')
signed_cert_file = os.path.join(tmpdir, "signed")
with open(signed_cert_file, 'w') as keyfile:
keyfile.write(cert_str)
project_cert_file = os.path.join(tmpdir, "project")
with open(project_cert_file, 'w') as keyfile:
keyfile.write(project_cert)
enc, err = utils.execute('openssl', 'verify', '-CAfile',
project_cert_file, '-verbose', signed_cert_file)
self.assertFalse(err)
开发者ID:B3n0n3，项目名称:nova，代码行数:19，代码来源:test_crypto.py
示例11: get_credentials
def get_credentials(self, user):
if not isinstance(user, User):
user = UserManager.instance().get_user(user)
rc = user.generate_rc(self.id)
private_key, signed_cert = self.generate_x509_cert(user)
tmpdir = tempfile.mkdtemp()
zf = os.path.join(tmpdir, "temp.zip")
zippy = zipfile.ZipFile(zf, 'w')
zippy.writestr(FLAGS.credential_rc_file, rc)
zippy.writestr(FLAGS.credential_key_file, private_key)
zippy.writestr(FLAGS.credential_cert_file, signed_cert)
zippy.writestr(FLAGS.ca_file, crypto.fetch_ca(self.id))
zippy.close()
with open(zf, 'rb') as f:
buffer = f.read()
shutil.rmtree(tmpdir)
return buffer
开发者ID:jxta，项目名称:cc，代码行数:19，代码来源:users.py
示例12: test_encrypt_decrypt_x509
def test_encrypt_decrypt_x509(self):
tmpdir = tempfile.mkdtemp()
self.flags(ca_path=tmpdir)
project_id = "fake"
try:
crypto.ensure_ca_filesystem()
cert = crypto.fetch_ca(project_id)
public_key = os.path.join(tmpdir, "public.pem")
with open(public_key, 'w') as keyfile:
keyfile.write(cert)
text = "some @#!%^* test text"
enc, _err = utils.execute('openssl',
'rsautl',
'-certin',
'-encrypt',
'-inkey', '%s' % public_key,
process_input=text)
dec = crypto.decrypt_text(project_id, enc)
self.assertEqual(text, dec)
finally:
shutil.rmtree(tmpdir)
开发者ID:acomisario，项目名称:nova，代码行数:21，代码来源:test_crypto.py
示例13: get_credentials
def get_credentials(self, user, project=None):
"""Get credential zip for user in project"""
if not isinstance(user, User):
user = self.get_user(user)
if project is None:
project = user.id
pid = Project.safe_id(project)
rc = self.__generate_rc(user.access, user.secret, pid)
private_key, signed_cert = self._generate_x509_cert(user.id, pid)
vpn = Vpn.lookup(pid)
if not vpn:
raise exception.Error("No vpn data allocated for project %s" %
project.name)
configfile = open(FLAGS.vpn_client_template,"r")
s = string.Template(configfile.read())
configfile.close()
config = s.substitute(keyfile=FLAGS.credential_key_file,
certfile=FLAGS.credential_cert_file,
ip=vpn.ip,
port=vpn.port)
tmpdir = tempfile.mkdtemp()
zf = os.path.join(tmpdir, "temp.zip")
zippy = zipfile.ZipFile(zf, 'w')
zippy.writestr(FLAGS.credential_rc_file, rc)
zippy.writestr(FLAGS.credential_key_file, private_key)
zippy.writestr(FLAGS.credential_cert_file, signed_cert)
zippy.writestr("nebula-client.conf", config)
zippy.writestr(FLAGS.ca_file, crypto.fetch_ca(user.id))
zippy.close()
with open(zf, 'rb') as f:
buffer = f.read()
shutil.rmtree(tmpdir)
return buffer
开发者ID:ChristopherMacGown，项目名称:my-nova，代码行数:36，代码来源:manager.py
示例14: send_root_ca
def send_root_ca(self):
username = self.get_username_from_ip(self.request.remote_ip)
self.set_header("Content-Type", "text/plain")
self.write(crypto.fetch_ca(username))
开发者ID:joshuamckenty，项目名称:pinet，代码行数:4，代码来源:api.py
示例15: fetch_ca
def fetch_ca(self, context, project_id):
"""Get root ca for a project."""
return crypto.fetch_ca(project_id)
开发者ID:674009287，项目名称:nova，代码行数:3，代码来源:manager.py
示例16: send_root_ca
def send_root_ca(self):
_log.debug( "Getting root ca")
project_id = self.get_project_id_from_ip(self.request.remote_ip)
self.set_header("Content-Type", "text/plain")
self.write(crypto.fetch_ca(project_id))
开发者ID:sorenh，项目名称:cc，代码行数:5，代码来源:api.py
示例17: fetch_crl
def fetch_crl(self, context, project_id):
"""Get crl for a project"""
return crypto.fetch_ca(project_id)
开发者ID:rcbops，项目名称:nova-buildpackage，代码行数:3，代码来源:manager.py
注：本文中的nova.crypto.fetch_ca函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论