我们从Python开源项目中，提取了以下11个代码示例，用于说明如何使用OpenSSL.SSL.VERIFY_FAIL_IF_NO_PEER_CERT。
def _server(self, sock): """ Create a new server-side SSL L{Connection} object wrapped around C{sock}. """ # Create the server side Connection. This is mostly setup boilerplate # - use TLSv1, use a particular certificate, etc. server_ctx = Context(TLSv1_METHOD) server_ctx.set_options(OP_NO_SSLv2 | OP_NO_SSLv3 | OP_SINGLE_DH_USE ) server_ctx.set_verify(VERIFY_PEER|VERIFY_FAIL_IF_NO_PEER_CERT|VERIFY_CLIENT_ONCE, verify_cb) server_store = server_ctx.get_cert_store() server_ctx.use_privatekey(load_privatekey(FILETYPE_PEM, server_key_pem)) server_ctx.use_certificate(load_certificate(FILETYPE_PEM, server_cert_pem)) server_ctx.check_privatekey() server_store.add_cert(load_certificate(FILETYPE_PEM, root_cert_pem)) # Here the Connection is actually created. If None is passed as the 2nd # parameter, it indicates a memory BIO should be created. server_conn = Connection(server_ctx, sock) server_conn.set_accept_state() return server_conn
def _client(self, sock): """ Create a new client-side SSL L{Connection} object wrapped around C{sock}. """ # Now create the client side Connection. Similar boilerplate to the # above. client_ctx = Context(TLSv1_METHOD) client_ctx.set_options(OP_NO_SSLv2 | OP_NO_SSLv3 | OP_SINGLE_DH_USE ) client_ctx.set_verify(VERIFY_PEER|VERIFY_FAIL_IF_NO_PEER_CERT|VERIFY_CLIENT_ONCE, verify_cb) client_store = client_ctx.get_cert_store() client_ctx.use_privatekey(load_privatekey(FILETYPE_PEM, client_key_pem)) client_ctx.use_certificate(load_certificate(FILETYPE_PEM, client_cert_pem)) client_ctx.check_privatekey() client_store.add_cert(load_certificate(FILETYPE_PEM, root_cert_pem)) client_conn = Connection(client_ctx, sock) client_conn.set_connect_state() return client_conn
def create_dest_sslcontext(insecure=False, trusted_ca_certs="", alpn=None): ssl_ctx = create_basic_sslcontext() if not insecure: trusted_ca_certs = trusted_ca_certs or certifi.where() ssl_ctx.load_verify_locations(trusted_ca_certs) ssl_ctx.set_verify( SSL.VERIFY_PEER | SSL.VERIFY_FAIL_IF_NO_PEER_CERT, certificate_verify_cb) else: ssl_ctx.set_verify(SSL.VERIFY_NONE, certificate_verify_cb) if alpn and HAS_ALPN: ssl_ctx.set_alpn_protos(alpn) return ssl_ctx
def _makeContext(self): ctx = SSL.Context(self.method) ctx.set_app_data(_SSLApplicationData()) if self.certificate is not None and self.privateKey is not None: ctx.use_certificate(self.certificate) ctx.use_privatekey(self.privateKey) # Sanity check ctx.check_privatekey() verifyFlags = SSL.VERIFY_NONE if self.verify: verifyFlags = SSL.VERIFY_PEER | SSL.VERIFY_FAIL_IF_NO_PEER_CERT if self.caCerts: store = ctx.get_cert_store() for cert in self.caCerts: store.add_cert(cert) def _trackVerificationProblems(conn,cert,errno,depth,preverify_ok): return True ctx.set_verify(verifyFlags, _trackVerificationProblems) if self.enableSessions: sessionName = md5.md5("%s-%d" % (reflect.qual(self.__class__), _sessionCounter())).hexdigest() ctx.set_session_id(sessionName) return ctx
def ssl_context(cacert, srvcrt, srvkey): # general setup: TLSv1.2, no compression, paranoid ciphers sslctx = SSL.Context(SSL.TLSv1_2_METHOD) sslctx.set_verify_depth(9) sslctx.set_options(SSL.OP_NO_COMPRESSION) sslctx.set_mode(_ssl_lib.SSL_MODE_ENABLE_PARTIAL_WRITE | _ssl_lib.SSL_MODE_ACCEPT_MOVING_WRITE_BUFFER) sslctx.set_cipher_list(libmu.defs.Defs.cipher_list) sslctx.set_verify(SSL.VERIFY_PEER | SSL.VERIFY_FAIL_IF_NO_PEER_CERT, lambda _, __, ___, ____, ok: ok) # use CA cert provided during lambda invocation fmt_cert = format_ssl_cert(cacert) x509_cert = crypto.load_certificate(crypto.FILETYPE_PEM, fmt_cert) sslctx.get_cert_store().add_cert(x509_cert) # add my certificate chain has_cert = False for cert in srvcrt.split(' '): x509_cert = crypto.load_certificate(crypto.FILETYPE_PEM, format_ssl_cert(cert)) if not has_cert: sslctx.use_certificate(x509_cert) has_cert = True else: sslctx.add_extra_chain_cert(x509_cert) # private key sslctx.use_privatekey(crypto.load_privatekey(crypto.FILETYPE_PEM, format_ssl_key(srvkey))) # check that all's well sslctx.check_privatekey() return sslctx ### # SSLize a connected socket, requiring a supplied cacert ###
def _makeContext(self): ctx = SSL.Context(self.method) ctx.set_app_data(_SSLApplicationData()) if self.certificate is not None and self.privateKey is not None: ctx.use_certificate(self.certificate) ctx.use_privatekey(self.privateKey) # Sanity check ctx.check_privatekey() verifyFlags = SSL.VERIFY_NONE if self.verify: verifyFlags = SSL.VERIFY_PEER if self.requireCertificate: verifyFlags |= SSL.VERIFY_FAIL_IF_NO_PEER_CERT if self.verifyOnce: verifyFlags |= SSL.VERIFY_CLIENT_ONCE if self.caCerts: store = ctx.get_cert_store() for cert in self.caCerts: store.add_cert(cert) def _trackVerificationProblems(conn,cert,errno,depth,preverify_ok): # retcode is the answer OpenSSL's default verifier would have # given, had we allowed it to run. if not preverify_ok: ctx.get_app_data().problems.append(OpenSSLVerifyError(cert, errno, depth)) return preverify_ok ctx.set_verify(verifyFlags, _trackVerificationProblems) if self.verifyDepth is not None: ctx.set_verify_depth(self.verifyDepth) if self.enableSingleUseKeys: ctx.set_options(SSL.OP_SINGLE_DH_USE) if self.fixBrokenPeers: ctx.set_options(self._OP_ALL) if self.enableSessions: sessionName = md5.md5("%s-%d" % (reflect.qual(self.__class__), _sessionCounter())).hexdigest() ctx.set_session_id(sessionName) return ctx
def _makeContext(self): ctx = self._contextFactory(self.method) ctx.set_options(self._options) ctx.set_mode(self._mode) if self.certificate is not None and self.privateKey is not None: ctx.use_certificate(self.certificate) ctx.use_privatekey(self.privateKey) for extraCert in self.extraCertChain: ctx.add_extra_chain_cert(extraCert) # Sanity check ctx.check_privatekey() verifyFlags = SSL.VERIFY_NONE if self.verify: verifyFlags = SSL.VERIFY_PEER if self.requireCertificate: verifyFlags |= SSL.VERIFY_FAIL_IF_NO_PEER_CERT if self.verifyOnce: verifyFlags |= SSL.VERIFY_CLIENT_ONCE self.trustRoot._addCACertsToContext(ctx) # It'd be nice if pyOpenSSL let us pass None here for this behavior (as # the underlying OpenSSL API call allows NULL to be passed). It # doesn't, so we'll supply a function which does the same thing. def _verifyCallback(conn, cert, errno, depth, preverify_ok): return preverify_ok ctx.set_verify(verifyFlags, _verifyCallback) if self.verifyDepth is not None: ctx.set_verify_depth(self.verifyDepth) if self.enableSessions: name = "%s-%d" % (reflect.qual(self.__class__), _sessionCounter()) sessionName = md5(networkString(name)).hexdigest() ctx.set_session_id(sessionName.encode('ascii')) if self.dhParameters: ctx.load_tmp_dh(self.dhParameters._dhFile.path) ctx.set_cipher_list(self._cipherString.encode('ascii')) if self._ecCurve is not None: try: self._ecCurve.addECKeyToContext(ctx) except BaseException: pass # ECDHE support is best effort only. if self._acceptableProtocols: # Try to set NPN and ALPN. _acceptableProtocols cannot be set by # the constructor unless at least one mechanism is supported. _setAcceptableProtocols(ctx, self._acceptableProtocols) return ctx
