Bommels05/Mozilla
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

commit for release 0.12, see doc/Changelog for details.

Browse Source 
Major enhancement is PKCS12 support.


git-svn-id: svn://10.0.0.236/trunk@262367 18797224-902f-48f8-a5cc-f745e15eee43
This commit is contained in:
jdennis%redhat.com
2011-06-06 15:21:13 +00:00
parent b99195930c
commit caf62c0df9
20 changed files with 4952 additions and 723 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
mozilla/security/python/nss/MANIFEST
View File

@@ -9,10 +9,13 @@ doc/examples/cert_dump.py
doc/examples/httplib_example.py
doc/examples/ssl_example.py
doc/examples/verify_server.py
test/cipher_test.py
test/digest_test.py
test/run_tests
test/setup_certs.py
test/test_cert_components.py
test/test_cipher.py
test/test_client_server.py
test/test_digest.py
test/test_pkcs12.py
src/NSPRerrs.h
src/SECerrs.h
src/SSLerrs.h

133
mozilla/security/python/nss/doc/ChangeLog
View File

@@ -1,3 +1,130 @@
2011-04-22 John Dennis <jdennis@redhat.com> 0.12
* Major new enhancement is additon of PKCS12 support and
AlgorithmID's.
* setup.py build enhancements
- Now searches for the NSS and NSPR header files rather
than hardcoding their location. This makes building friendlier
on other systems (i.e. debian)
- Now takes optional command line arguments, -d or --debug
will turn on debug options during the build.
* Fix reference counting bug in PK11_password_callback() which
contributed to NSS not being able to shutdown due to
resources still in use.
* Add UTF-8 support to ssl.config_server_session_id_cache()
* Added unit tests for cipher, digest, client_server.
* All unittests now run, added test/run_tests to invoke
full test suite.
* Fix bug in test/setup_certs.py, hardcoded full path to
libnssckbi.so was causing failures on 64-bit systems,
just use the libnssckbi.so basename, modutil will find
it on the standard search path.
* doc/examples/cert_dump.py uses new AlgorithmID class to
dump Signature Algorithm
* doc/examples/ssl_example.py now can cleanly shutdown NSS.
* Exception error messages now include PR error text if available.
* The following classes were replaced:
- SignatureAlgorithm replaced by new class AlgorithmID
* The following classes were added:
- AlgorithmID
- PKCS12DecodeItem
- PKCS12Decoder
* The following class methods were added:
- PK11Slot.authenticate()
- PK11Slot.get_disabled_reason()
- PK11Slot.has_protected_authentication_path()
- PK11Slot.has_root_certs()
- PK11Slot.is_disabled()
- PK11Slot.is_friendly()
- PK11Slot.is_internal()
- PK11Slot.is_logged_in()
- PK11Slot.is_removable()
- PK11Slot.logout()
- PK11Slot.need_login()
- PK11Slot.need_user_init()
- PK11Slot.user_disable()
- PK11Slot.user_enable()
- PKCS12DecodeItem.format()
- PKCS12DecodeItem.format_lines()
- PKCS12Decoder.database_import()
- PKCS12Decoder.format()
- PKCS12Decoder.format_lines()
* The following class properties were added:
- AlgorithmID.id_oid
- AlgorithmID.id_str
- AlgorithmID.id_tag
- AlgorithmID.parameters
- PKCS12DecodeItem.certificate
- PKCS12DecodeItem.friendly_name
- PKCS12DecodeItem.has_key
- PKCS12DecodeItem.shroud_algorithm_id
- PKCS12DecodeItem.signed_cert_der
- PKCS12DecodeItem.type
- SignedData.data
- SignedData.der
* The following module functions were added:
- nss.nss.dump_certificate_cache_info()
- nss.nss.find_slot_by_name()
- nss.nss.fingerprint_format_lines()
- nss.nss.get_internal_slot()
- nss.nss.is_fips()
- nss.nss.need_pw_init()
- nss.nss.nss_init_read_write()
- nss.nss.pk11_disabled_reason_name()
- nss.nss.pk11_disabled_reason_str()
- nss.nss.pk11_logout_all()
- nss.nss.pkcs12_cipher_from_name()
- nss.nss.pkcs12_cipher_name()
- nss.nss.pkcs12_enable_all_ciphers()
- nss.nss.pkcs12_enable_cipher()
- nss.nss.pkcs12_export()
- nss.nss.pkcs12_map_cipher()
- nss.nss.pkcs12_set_nickname_collision_callback()
- nss.nss.pkcs12_set_preferred_cipher()
- nss.nss.token_exists()
- nss.ssl.config_mp_server_sid_cache()
- nss.ssl.config_server_session_id_cache_with_opt()
- nss.ssl.get_max_server_cache_locks()
- nss.ssl.set_max_server_cache_locks()
- nss.ssl.shutdown_server_session_id_cache()
* The following constants were added:
- nss.nss.int.PK11_DIS_COULD_NOT_INIT_TOKEN
- nss.nss.int.PK11_DIS_NONE
- nss.nss.int.PK11_DIS_TOKEN_NOT_PRESENT
- nss.nss.int.PK11_DIS_TOKEN_VERIFY_FAILED
- nss.nss.int.PK11_DIS_USER_SELECTED
- nss.nss.int.PKCS12_DES_56
- nss.nss.int.PKCS12_DES_EDE3_168
- nss.nss.int.PKCS12_RC2_CBC_128
- nss.nss.int.PKCS12_RC2_CBC_40
- nss.nss.int.PKCS12_RC4_128
- nss.nss.int.PKCS12_RC4_40
* The following files were added:
- test/run_tests
- test/test_cipher.py (replaces cipher_test.py)
- test/test_client_server.py
- test/test_digest.py (replaces digest_test.py)
- test/test_pkcs12.py
* The following were deprecated:
- SignatureAlgorithm
2011-02-21 John Dennis <jdennis@redhat.com> 0.11
External Changes:
@@ -151,7 +278,7 @@
* The following classes were added:
AuthKeyID
BasicConstraints
BasicConstraints
CRLDistributionPoint
CRLDistributionPts
CertificateExtension
@@ -161,7 +288,7 @@
RDN
AVA
CertificateRequest
* The following module functions were added:
nss.nss.nss_is_initialized()
nss.nss.cert_crl_reason_from_name()
@@ -247,7 +374,7 @@
adding object traversal and clear methods
* Identify static variables, move to thread local storage
* Remove python-nss specific httplib.py, no longer needed
python-nss now compatible with standard library

3
mozilla/security/python/nss/doc/examples/cert_dump.py
View File

@@ -167,7 +167,8 @@ print nss.indented_format([(0, 'Certificate:'),
(1, 'Data:')])
print nss.indented_format([(2, 'Version: %d (%#x)' % (cert.version+1, cert.version))])
print nss.indented_format([(2, 'Serial Number: %d (%#x)' % (cert.serial_number, cert.serial_number))])
print nss.indented_format([(2, 'Signature Algorithm: %s' % cert.signature_algorithm)])
print nss.indented_format([(2, 'Signature Algorithm:')])
print nss.indented_format(cert.signature_algorithm.format_lines(3))
print nss.indented_format([(2, 'Issuer: "%s"' % cert.issuer)])
print nss.indented_format([(2, 'Validity:'),
(3, 'Not Before: %s' % cert.valid_not_before_str),

38
mozilla/security/python/nss/doc/examples/ssl_example.py
View File

@@ -60,7 +60,7 @@ REQUIRE_CLIENT_CERT_ALWAYS = 4
# command line parameters, default them to something reasonable
client = False
server = False
password = 'passwd'
password = 'db_passwd'
use_ssl = True
client_cert_action = NO_CLIENT_CERT
certdir = 'pki'
@@ -199,7 +199,6 @@ def Client():
sock.set_ssl_option(ssl.SSL_SECURITY, True)
sock.set_ssl_option(ssl.SSL_HANDSHAKE_AS_CLIENT, True)
sock.set_hostname(hostname)
sock.reset_handshake(False) # FIXME: is this needed
# Provide a callback which notifies us when the SSL handshake is complete
sock.set_handshake_callback(handshake_callback)
@@ -221,22 +220,13 @@ def Client():
valid_addr = True
break
except Exception, e:
sock.close()
print "client connection to: %s failed (%s)" % (net_addr, e)
if not valid_addr:
print "Could not establish valid address for \"%s\" in family %s" % \
(hostname, io.addr_family_name(family))
return
try:
print "client connected to: %s" % sock.get_peer_name()
except Exception, e:
print e.strerror
try:
sock.shutdown()
except:
pass
return
# Talk to the server
try:
@@ -244,12 +234,13 @@ def Client():
buf = sock.recv(1024)
if not buf:
print "client lost connection"
sock.close()
return
print "client received: %s" % (buf)
except Exception, e:
print e.strerror
try:
sock.shutdown()
sock.close()
except:
pass
return
@@ -260,7 +251,13 @@ def Client():
sock.shutdown()
except:
pass
return
try:
sock.close()
if use_ssl:
ssl.clear_session_cache()
except Exception, e:
print e
# -----------------------------------------------------------------------------
# Server Implementation
@@ -302,7 +299,6 @@ def Server():
# Configure the server SSL socket
sock.config_secure_server(server_cert, priv_key, server_cert_kea)
sock.reset_handshake(True) # FIXME: is this needed?
else:
sock = io.Socket(net_addr.family)
@@ -333,6 +329,7 @@ def Server():
client_sock.send("Goodbye")
try:
client_sock.shutdown(io.PR_SHUTDOWN_RCV)
client_sock.close()
except:
pass
break
@@ -343,7 +340,11 @@ def Server():
try:
sock.shutdown()
except:
sock.close()
if use_ssl:
ssl.shutdown_server_session_id_cache()
except Exception, e:
print e
pass
# -----------------------------------------------------------------------------
@@ -486,3 +487,8 @@ if server:
print "starting as server"
Server()
try:
nss.nss_shutdown()
except Exception, e:
print e

58
mozilla/security/python/nss/doc/get_api Executable file
View File

@@ -0,0 +1,58 @@
#!/usr/bin/python
import types
import inspect
import nss.io as io
import nss.ssl as ssl
import nss.nss as nss
def isint(object):
"Return true if the object is an integer."
return isinstance(object, types.IntType)
def isstring(object):
"Return true if the object is a string."
return isinstance(object, types.StringTypes)
class Package:
def __init__(self, modules):
self.modules = modules
class Class:
def __init__(self, cls):
self.cls = cls
self.name = cls.__name__
self.methods = [x[0] for x in inspect.getmembers(self.cls, inspect.ismethoddescriptor) if not x[0].startswith('_')]
self.properties = [x[0] for x in inspect.getmembers(cls, inspect.isgetsetdescriptor)]
class Module:
def __init__(self, module):
self.module = module
self.name = module.__name__
self.classes = [Class(x) for x in [x[1] for x in inspect.getmembers(self.module, inspect.isclass)]]
self.functions = [x[0] for x in inspect.getmembers(self.module, inspect.isbuiltin)]
self.ints = [x[0] for x in inspect.getmembers(self.module, isint)]
self.strings = [x[0] for x in inspect.getmembers(self.module, isstring)]
def dump(self):
for cls in self.classes:
print '%s.class.%s' % (self.name, cls.name)
for prop in cls.properties:
print '%s.property.%s.%s' % (self.name, cls.name, prop)
for method in cls.methods:
print '%s.method.%s.%s()' % (self.name, cls.name, method)
for func in self.functions:
print "%s.function.%s()" % (self.name, func)
for i in self.ints:
print "%s.int.%s" % (self.name, i)
for s in self.strings:
print "%s.str.%s" % (self.name, s)
for m in (io, ssl, nss):
mod = Module(m)
mod.dump()

198
mozilla/security/python/nss/setup.py
View File

@@ -50,7 +50,19 @@ from distutils.util import subst_vars, change_root
from distutils.command.build_py import build_py as _build_py
from distutils.command.sdist import sdist as _sdist
version = "0.11"
version = "0.12"
doc_manifest = [
[['include README LICENSE* doc/ChangeLog',
'recursive-include doc *.py *.txt',],
[('^doc/', '')], None],
[['recursive-include test *.py *.txt',],
None , None],
[['recursive-include lib *.py *.txt',],
[('^lib/', '')] , 'examples'],
[['recursive-include build/doc/html *'],
[('^build/doc/', 'api/')], None],
]
def update_version():
"""If the version string in __init__.py doesn't match the current
@@ -83,6 +95,33 @@ def update_version():
else:
os.unlink(tmp_file)
def find_include_dir(dir_names, include_files, include_roots=['/usr/include', '/usr/local/include']):
'''
Locate an include directory on the system which contains the specified include files.
You must provide a list of directory basenames to search. You may optionally provide
a list of include roots. The search proceeds by iterating over each root and appending
each directory basename to it. If the resulting directory path contains all the include
files that directory is returned. If no directory is found containing all the include
files a ValueError is raised.
'''
if len(dir_names) == 0:
raise ValueError("directory search list is empty")
if len(include_files) == 0:
raise ValueError("header file list is empty")
for include_root in include_roots:
for dir_name in dir_names:
include_dir = os.path.join(include_root, dir_name)
if os.path.isdir(include_dir):
for include_file in include_files:
found = True
file_path = os.path.join(include_dir, include_file)
if not os.path.exists(file_path):
found = False
break
if found:
return include_dir
raise ValueError("unable to locate include directory containing header files %s" % include_files)
class BuildPy(_build_py):
"""Specialized Python source builder."""
@@ -298,89 +337,92 @@ class InstallDoc(Command):
#------------------------------------------------------------------------------
long_description='''
Python bindings for Network Security Services (NSS) and Netscape Portable Runtime (NSPR).
'''
def main(argv):
debug_compile_args = ['-O0', '-g']
# force debug build until package matures
#extra_compile_args = debug_compile_args
extra_compile_args = []
long_description='''
Python bindings for Network Security Services (NSS) and Netscape Portable Runtime (NSPR).
'''
nss_error_extension = \
Extension('nss.error',
sources = ['src/py_nspr_error.c'],
include_dirs = ['/usr/include/nss3', '/usr/include/nspr4'],
depends = ['src/py_nspr_common.h', 'src/py_nspr_error.h'],
libraries = ['nspr4'],
extra_compile_args = extra_compile_args,
)
debug_compile_args = ['-O0', '-g']
extra_compile_args = []
nss_io_extension = \
Extension('nss.io',
sources = ['src/py_nspr_io.c'],
include_dirs = ['/usr/include/nss3', '/usr/include/nspr4'],
depends = ['src/py_nspr_common.h', 'src/py_nspr_error.h', 'src/py_nspr_io.h'],
libraries = ['nspr4'],
extra_compile_args = extra_compile_args,
)
for arg in argv:
if arg in ('-d', '--debug'):
print "compiling with debug"
extra_compile_args = debug_compile_args
argv.remove(arg)
nss_nss_extension = \
Extension('nss.nss',
sources = ['src/py_nss.c'],
include_dirs = ['src', '/usr/include/nss3', '/usr/include/nspr4'],
depends = ['src/py_nspr_common.h', 'src/py_nspr_error.h', 'src/py_nss.h'],
libraries = ['nspr4', 'ssl3', 'nss3'],
extra_compile_args = extra_compile_args,
)
nss_include_dir = find_include_dir(['nss3', 'nss'], ['nss.h', 'pk11pub.h'])
nspr_include_dir = find_include_dir(['nspr4', 'nspr'], ['nspr.h', 'prio.h'])
nss_ssl_extension = \
Extension('nss.ssl',
sources = ['src/py_ssl.c'],
include_dirs = ['src', '/usr/include/nss3', '/usr/include/nspr4'],
depends = ['src/py_nspr_common.h', 'src/py_nspr_error.h', 'src/py_nspr_io.h',
'src/py_ssl.h', 'src/py_nss.h'],
libraries = ['nspr4', 'ssl3'],
extra_compile_args = extra_compile_args,
)
nss_error_extension = \
Extension('nss.error',
sources = ['src/py_nspr_error.c'],
include_dirs = [nss_include_dir, nspr_include_dir],
depends = ['src/py_nspr_common.h', 'src/py_nspr_error.h'],
libraries = ['nspr4'],
extra_compile_args = extra_compile_args,
)
doc_manifest = [
[['include README LICENSE* doc/ChangeLog',
'recursive-include doc *.py *.txt',],
[('^doc/', '')], None],
[['recursive-include test *.py *.txt',],
None , None],
[['recursive-include lib *.py *.txt',],
[('^lib/', '')] , 'examples'],
[['recursive-include build/doc/html *'],
[('^build/doc/', 'api/')], None],
]
nss_io_extension = \
Extension('nss.io',
sources = ['src/py_nspr_io.c'],
include_dirs = [nss_include_dir, nspr_include_dir],
depends = ['src/py_nspr_common.h', 'src/py_nspr_error.h', 'src/py_nspr_io.h'],
libraries = ['nspr4'],
extra_compile_args = extra_compile_args,
)
setup(name = 'python-nss',
version = version,
description = 'Python bindings for Network Security Services (NSS) and Netscape Portable Runtime (NSPR)',
long_description = long_description,
author = 'John Dennis',
author_email = 'jdennis@redhat.com',
maintainer = 'John Dennis',
maintainer_email = 'jdennis@redhat.com',
license = 'MPLv1.1 or GPLv2+ or LGPLv2+',
platforms = 'posix',
url = ':pserver:anonymous@cvs-mirror.mozilla.org:/cvsroot mozilla/security/python/nss',
download_url = '',
ext_modules = [nss_error_extension,
nss_io_extension,
nss_nss_extension,
nss_ssl_extension,
],
package_dir = {'nss':'src'},
packages = ['nss'],
cmdclass = {'build_doc' : BuildDoc,
'build_api_doc' : BuildApiDoc,
'install_doc' : InstallDoc,
'build_py' : BuildPy,
'sdist' : SDist,
},
nss_nss_extension = \
Extension('nss.nss',
sources = ['src/py_nss.c'],
include_dirs = ['src', nss_include_dir, nspr_include_dir],
depends = ['src/py_nspr_common.h', 'src/py_nspr_error.h', 'src/py_nss.h'],
libraries = ['nspr4', 'ssl3', 'nss3', 'smime3'],
extra_compile_args = extra_compile_args,
)
)
nss_ssl_extension = \
Extension('nss.ssl',
sources = ['src/py_ssl.c'],
include_dirs = ['src', nss_include_dir, nspr_include_dir],
depends = ['src/py_nspr_common.h', 'src/py_nspr_error.h', 'src/py_nspr_io.h',
'src/py_ssl.h', 'src/py_nss.h'],
libraries = ['nspr4', 'ssl3'],
extra_compile_args = extra_compile_args,
)
setup(name = 'python-nss',
version = version,
description = 'Python bindings for Network Security Services (NSS) and Netscape Portable Runtime (NSPR)',
long_description = long_description,
author = 'John Dennis',
author_email = 'jdennis@redhat.com',
maintainer = 'John Dennis',
maintainer_email = 'jdennis@redhat.com',
license = 'MPLv1.1 or GPLv2+ or LGPLv2+',
platforms = 'posix',
url = ':pserver:anonymous@cvs-mirror.mozilla.org:/cvsroot mozilla/security/python/nss',
download_url = '',
ext_modules = [nss_error_extension,
nss_io_extension,
nss_nss_extension,
nss_ssl_extension,
],
package_dir = {'nss':'src'},
packages = ['nss'],
cmdclass = {'build_doc' : BuildDoc,
'build_api_doc' : BuildApiDoc,
'install_doc' : InstallDoc,
'build_py' : BuildPy,
'sdist' : SDist,
},
)
return 0
#------------------------------------------------------------------------------
if __name__ == "__main__":
sys.exit(main(sys.argv))

2
mozilla/security/python/nss/src/__init__.py
View File

@@ -345,5 +345,5 @@ FAQ
To be added
"""
__version__ = '0.11'
__version__ = '0.12'

16
mozilla/security/python/nss/src/py_nspr_common.h
View File

@@ -47,6 +47,8 @@
#define NSS_THREAD_LOCAL_KEY "nss"
#define PyBoolAsPRBool(x) ((x) == Py_True ? PR_TRUE : PR_FALSE)
#define ASSIGN_REF(dst, obj) \
do { \
PyObject *tmp; \
@@ -97,6 +99,8 @@ typedef PyObject *(*ssizessizeargfunc)(PyObject *, Py_ssize_t, Py_ssize_t);
#define Py_TPFLAGS_HAVE_NEWBUFFER 0
#endif
#define PyNone_Check(x) ((x) == Py_None)
#define TYPE_READY(type) \
{ \
if (PyType_Ready(&type) < 0) \
@@ -109,8 +113,9 @@ typedef PyObject *(*ssizessizeargfunc)(PyObject *, Py_ssize_t, Py_ssize_t);
#ifdef DEBUG
#define DumpRefCount(_obj) \
#define DumpRefCount(x) \
{ \
PyObject *_obj = (PyObject *) (x); \
printf("<%s object at %p refcnt=%d>\n", Py_TYPE(_obj)->tp_name, _obj, _obj->ob_refcnt); \
}
@@ -120,8 +125,9 @@ typedef PyObject *(*ssizessizeargfunc)(PyObject *, Py_ssize_t, Py_ssize_t);
printf("%s\n", _msg); \
}
#define TraceMethodEnter(_obj) \
#define TraceMethodEnter(x) \
{ \
PyObject *_obj = (PyObject *) (x); \
const char *name = NULL; \
\
if (_obj) { \
@@ -131,8 +137,9 @@ typedef PyObject *(*ssizessizeargfunc)(PyObject *, Py_ssize_t, Py_ssize_t);
__FUNCTION__, name, _obj, _obj ? _obj->ob_refcnt : -9999); \
}
#define TraceMethodLeave(_obj) \
#define TraceMethodLeave(x) \
{ \
PyObject *_obj = (PyObject *) (x); \
const char *name = NULL; \
\
if (_obj) { \
@@ -152,8 +159,9 @@ typedef PyObject *(*ssizessizeargfunc)(PyObject *, Py_ssize_t, Py_ssize_t);
}
#define TraceObjNewLeave(_obj) \
#define TraceObjNewLeave(x) \
{ \
PyObject *_obj = (PyObject *) (x); \
const char *name = NULL; \
\
if (_obj) { \

40
mozilla/security/python/nss/src/py_nspr_error.c
View File

@@ -140,10 +140,11 @@ set_nspr_error(const char *format, ...)
va_list vargs;
PyObject *v;
const NSPRErrorDesc *error_desc;
char *errstr=NULL;
PRErrorCode err;
char *pr_err_msg=NULL;
PRInt32 pr_err_msg_len;
PRErrorCode error_code;
PyObject *detail = NULL;
char buf[1024];
char *final_err_msg=NULL;
if (format) {
#ifdef HAVE_STDARG_PROTOTYPES
@@ -155,28 +156,39 @@ set_nspr_error(const char *format, ...)
va_end(vargs);
}
err = PR_GetError();
PR_GetErrorText(errstr);
if (errstr == NULL) {
if ((error_desc = lookup_nspr_error(err)) != NULL) {
snprintf(buf, sizeof(buf), "(%s) %s", error_desc->name, error_desc->string);
errstr = buf;
} else {
errstr = NULL;
}
error_code = PR_GetError();
error_desc = lookup_nspr_error(error_code);
if ((pr_err_msg_len = PR_GetErrorTextLength())) {
if ((pr_err_msg = PyMem_Malloc(pr_err_msg_len + 1))) {
PR_GetErrorText(pr_err_msg);
}
}
if (pr_err_msg && error_desc) {
final_err_msg = PR_smprintf("%s (%s) %s", pr_err_msg, error_desc->name, error_desc->string);
} else if (error_desc) {
final_err_msg = PR_smprintf("(%s) %s", error_desc->name, error_desc->string);
} else if (pr_err_msg) {
final_err_msg = PR_smprintf("%s", pr_err_msg);
} else {
final_err_msg = PR_smprintf("error (%d) unknown", error_code);
}
if (detail) {
v = Py_BuildValue("(isS)", err, errstr, detail);
v = Py_BuildValue("(isS)", error_code, final_err_msg, detail);
Py_DECREF(detail);
} else {
v = Py_BuildValue("(is)", err, errstr);
v = Py_BuildValue("(is)", error_code, final_err_msg);
}
if (v != NULL) {
PyErr_SetObject(NSPR_Exception, v);
Py_DECREF(v);
}
if (final_err_msg) PR_smprintf_free(final_err_msg);
if (pr_err_msg) PyMem_Free(pr_err_msg);
return NULL;
}

3479
mozilla/security/python/nss/src/py_nss.c
View File

File diff suppressed because it is too large Load Diff

30
mozilla/security/python/nss/src/py_nss.h
View File

@@ -49,6 +49,7 @@
#include "sslt.h"
#include "key.h"
#include "pk11pub.h"
#include "pkcs12.h"
/* ========================================================================== */
@@ -185,7 +186,7 @@ typedef struct {
} KEYPQGParams;
/* ========================================================================== */
/* ======================== SignatureAlgorithm Class ======================== */
/* ============================ AlgorithmID Class =========================== */
/* ========================================================================== */
typedef struct {
@@ -193,7 +194,7 @@ typedef struct {
SECAlgorithmID id;
PyObject *py_id;
PyObject *py_parameters;
} SignatureAlgorithm;
} AlgorithmID;
/* ========================================================================== */
/* ============================= SignedData Class =========================== */
@@ -203,6 +204,7 @@ typedef struct {
PyObject_HEAD
PRArenaPool *arena;
CERTSignedData signed_data;
PyObject *py_der;
PyObject *py_data;
PyObject *py_algorithm;
PyObject *py_signature;
@@ -355,6 +357,30 @@ typedef struct {
NSSInitContext *context;
} InitContext;
/* ========================================================================== */
/* =========================== PKCS12DecodeItem Class ======================= */
/* ========================================================================== */
typedef struct {
PyObject_HEAD
SECOidTag type;
PRBool has_key;
PyObject *py_signed_cert_der;
PyObject *py_cert;
PyObject *py_friendly_name;
PyObject *py_shroud_algorithm_id;
} PKCS12DecodeItem;
/* ========================================================================== */
/* ============================ PKCS12Decoder Class ========================= */
/* ========================================================================== */
typedef struct {
PyObject_HEAD
SEC_PKCS12DecoderContext *decoder_ctx;
PyObject *py_decode_items; /* tuple */
} PKCS12Decoder;
/* ========================================================================== */
typedef struct {

359
mozilla/security/python/nss/src/py_ssl.c
View File

@@ -1,3 +1,30 @@
#if 0
PyDoc_STRVAR(SSL_func_name_doc,
"func_name() -> \n\
\n\
:Parameters:\n\
arg1 : object\n\
xxx\n\
\n\
xxx\n\
");
static PyObject *
SSL_func_name(PyObject *self, PyObject *args, PyObject *kwds)
{
static char *kwlist[] = {"arg1", NULL};
PyObject *arg;
TraceMethodEnter(self);
if (!PyArg_ParseTupleAndKeywords(args, kwds, "O|i:func_name", kwlist,
&arg))
return NULL;
return NULL;
}
#endif
/* ***** BEGIN LICENSE BLOCK *****
* Version: MPL 1.1/GPL 2.0/LGPL 2.1
*
@@ -1293,8 +1320,8 @@ PyDoc_STRVAR(SSLSocket_set_cipher_pref_doc,
:Parameters:\n\
cipher : integer\n\
The cipher suite enumeration (e.g. SSL_RSA_WITH_NULL_MD5, etc.)\n\
enabled : bool\n\
Boolean value\n\
enabled : bool or int\n\
True enables, False disables\n\
\n\
Sets preference for the specified SSL2, SSL3, or TLS cipher on the\n\
socket. A cipher suite is used only if the policy allows it and the\n\
@@ -1919,6 +1946,8 @@ NSSinit(PyObject *self, PyObject *args)
{
char *cert_dir;
TraceMethodEnter(self);
if (PyErr_Warn(PyExc_DeprecationWarning, "nssinit() has been moved to the nss module, use nss.nss_init() instead of ssl.nssinit()") < 0)
return NULL;
@@ -1950,6 +1979,8 @@ NSS_init(PyObject *self, PyObject *args)
{
char *cert_dir;
TraceMethodEnter(self);
if (PyErr_Warn(PyExc_DeprecationWarning, "nss_init() has been moved to the nss module, use nss.nss_init() instead of ssl.nss_init()") < 0)
return NULL;
@@ -1978,6 +2009,8 @@ with the error code SEC_ERROR_BUSY.\n\
static PyObject *
NSS_shutdown(PyObject *self, PyObject *args)
{
TraceMethodEnter(self);
if (PyErr_Warn(PyExc_DeprecationWarning, "nss_shutdown() has been moved to the nss module, use nss.nss_shutdown() instead of ssl.nss_shutdown()") < 0)
return NULL;
@@ -2006,6 +2039,8 @@ SSL_set_ssl_default_option(PyObject *self, PyObject *args)
int option;
int value;
TraceMethodEnter(self);
if (!PyArg_ParseTuple(args, "ii:set_ssl_default_option", &option, &value)) {
return NULL;
}
@@ -2031,6 +2066,8 @@ SSL_get_ssl_default_option(PyObject *self, PyObject *args)
int option;
int value;
TraceMethodEnter(self);
if (!PyArg_ParseTuple(args, "i:get_ssl_default_option", &option)) {
return NULL;
}
@@ -2068,6 +2105,8 @@ SSL_set_default_cipher_pref(PyObject *self, PyObject *args)
int cipher;
int enabled;
TraceMethodEnter(self);
if (!PyArg_ParseTuple(args, "ii:set_default_cipher_pref", &cipher, &enabled))
return NULL;
@@ -2095,6 +2134,8 @@ SSL_get_default_cipher_pref(PyObject *self, PyObject *args)
int cipher;
int enabled;
TraceMethodEnter(self);
if (!PyArg_ParseTuple(args, "i:get_default_cipher_pref", &cipher))
return NULL;
@@ -2135,6 +2176,8 @@ SSL_set_cipher_policy(PyObject *self, PyObject *args)
int cipher;
int policy;
TraceMethodEnter(self);
if (!PyArg_ParseTuple(args, "ii:set_cipher_policy", &cipher, &policy))
return NULL;
@@ -2161,6 +2204,8 @@ SSL_get_cipher_policy(PyObject *self, PyObject *args)
int cipher;
int policy;
TraceMethodEnter(self);
if (!PyArg_ParseTuple(args, "i:get_cipher_policy", &cipher))
return NULL;
@@ -2176,7 +2221,7 @@ SSL_get_cipher_policy(PyObject *self, PyObject *args)
}
PyDoc_STRVAR(SSL_config_server_session_id_cache_doc,
"config_server_session_id_cache([max_cache_entries=0, ssl2_timeout=0, ssl3_timeout=0, directory=None])\n\
"config_server_session_id_cache(max_cache_entries=0, ssl2_timeout=0, ssl3_timeout=0, directory=None)\n\
\n\
:Parameters:\n\
max_cache_entries : integer\n\
@@ -2203,7 +2248,7 @@ handshake as a server, you must call config_server_session_id_cache()\n\
to configure the session caches for server sessions.\n\
\n\
If your server application uses multiple processes (instead of or in\n\
addition to multiple threads), use config_mp_server_sid_cache()\n\
addition to multiple threads), use `ssl.config_mp_server_sid_cache()`\n\
instead. You must use one of these functions to create a server\n\
cache.\n\
\n\
@@ -2228,20 +2273,267 @@ SSL_config_server_session_id_cache(PyObject *self, PyObject *args, PyObject *kwd
int max_cache_entries = 0;
PRUint32 ssl2_timeout = 0;
PRUint32 ssl3_timeout = 0;
PyObject *py_directory = Py_None;
PyObject *py_directory_utf8 = NULL;
char *directory = NULL;
TraceMethodEnter(self);
if (!PyArg_ParseTupleAndKeywords(args, kwds, "|iIIz:config_server_session_id_cache", kwlist,
&max_cache_entries, &ssl2_timeout, &ssl3_timeout, &directory))
if (!PyArg_ParseTupleAndKeywords(args, kwds, "|iIIO:config_server_session_id_cache", kwlist,
&max_cache_entries, &ssl2_timeout, &ssl3_timeout, &py_directory))
return NULL;
if (PyString_Check(py_directory) || PyUnicode_Check(py_directory)) {
if (PyString_Check(py_directory)) {
py_directory_utf8 = py_directory;
Py_INCREF(py_directory_utf8);
} else {
py_directory_utf8 = PyUnicode_AsUTF8String(py_directory);
}
directory = PyString_AsString(py_directory_utf8);
} else if (PyNone_Check(py_directory)) {
directory = NULL;
} else {
PyErr_Format(PyExc_TypeError, "directory must be string or None, not %.200s",
Py_TYPE(py_directory)->tp_name);
return NULL;
}
if (SSL_ConfigServerSessionIDCache(max_cache_entries, ssl2_timeout,
ssl3_timeout, directory) != SECSuccess) {
Py_XDECREF(py_directory_utf8);
return set_nspr_error(NULL);
}
Py_XDECREF(py_directory_utf8);
Py_RETURN_NONE;
}
PyDoc_STRVAR(SSL_config_server_session_id_cache_with_opt_doc,
"config_server_session_id_cache_with_opt(max_cache_entries=0, max_cert_cache_entries=0, max_server_name_cache_entries=0, ssl2_timeout=0, ssl3_timeout=0, directory=None, enable_mp_cache=False)\n\
\n\
:Parameters:\n\
max_cache_entries : integer\n\
The maximum number of entries in the cache. If ZERO the server\n\
default value is used (10,000).\n\
\n\
max_cert_cache_entries : integer\n\
The maximum number of entries in the cert cache. If ZERO the server\n\
default value is used (10,000).\n\
\n\
max_server_name_cache_entries : integer\n\
The maximum number of entries in the server name cache. If ZERO the server\n\
default value is used (10,000).\n\
\n\
ssl2_timeout : integer\n\
The lifetime in seconds of an SSL2 session. The minimum timeout\n\
value is 5 seconds and the maximum is 24 hours. Values outside\n\
this range are replaced by the server default value (100 seconds).\n\
\n\
ssl3_timeout : integer\n\
The lifetime in seconds of an SSL3 session. The minimum timeout\n\
value is 5 seconds and the maximum is 24 hours. Values outside\n\
this range are replaced by the server default value (24 hours).\n\
\n\
directory : string\n\
A string specifying the pathname of the directory that will\n\
contain the session cache. If None the server default value is\n\
used (/tmp (Unix) or \\temp (NT)).\n\
\n\
enable_mp_cache : bool\n\
If True enable the multi-process cache.\n\
\n\
Configure a secure server's session-id cache. Depends on value of\n\
enable_mp_cache, configures multi-proc or single proc cache.\n\
\n\
A zero value or a value that is out of range for any of the parameters\n\
causes the server default value to be used in the server cache. Note,\n\
this function only affects the server cache, not the client cache.\n\
");
static PyObject *
SSL_config_server_session_id_cache_with_opt(PyObject *self, PyObject *args, PyObject *kwds)
{
static char *kwlist[] = {"max_cache_entries", "max_cert_cache_entries", "max_server_name_cache_entries",
"ssl2_timeout", "ssl3_timeout", "directory", "enable_mp_cache", NULL};
int max_cache_entries = 0;
int max_cert_cache_entries = 0;
int max_server_name_cache_entries = 0;
PRUint32 ssl2_timeout = 0;
PRUint32 ssl3_timeout = 0;
PyObject *py_directory = Py_None;
PyObject *py_directory_utf8 = NULL;
char *directory = NULL;
PyObject * py_enable_mp_cache = NULL;
PRBool enable_mp_cache = PR_FALSE;
TraceMethodEnter(self);
if (!PyArg_ParseTupleAndKeywords(args, kwds, "|iiiIIO:config_server_session_id_cache_with_opt", kwlist,
&max_cache_entries, &max_cert_cache_entries, &max_server_name_cache_entries,
&ssl2_timeout, &ssl3_timeout, &py_directory, &py_enable_mp_cache))
return NULL;
if (PyString_Check(py_directory) || PyUnicode_Check(py_directory)) {
if (PyString_Check(py_directory)) {
py_directory_utf8 = py_directory;
Py_INCREF(py_directory_utf8);
} else {
py_directory_utf8 = PyUnicode_AsUTF8String(py_directory);
}
directory = PyString_AsString(py_directory_utf8);
} else if (PyNone_Check(py_directory)) {
directory = NULL;
} else {
PyErr_Format(PyExc_TypeError, "directory must be string or None, not %.200s",
Py_TYPE(py_directory)->tp_name);
return NULL;
}
if (py_enable_mp_cache) {
enable_mp_cache = PyBoolAsPRBool(py_enable_mp_cache);
}
if (SSL_ConfigServerSessionIDCacheWithOpt(ssl2_timeout, ssl3_timeout, directory,
max_cache_entries, max_cert_cache_entries,
max_server_name_cache_entries,
enable_mp_cache) != SECSuccess) {
Py_XDECREF(py_directory_utf8);
return set_nspr_error(NULL);
}
Py_XDECREF(py_directory_utf8);
Py_RETURN_NONE;
}
PyDoc_STRVAR(SSL_config_mp_server_sid_cache_doc,
"config_mp_server_sid_cache(max_cache_entries=0, ssl2_timeout=0, ssl3_timeout=0, directory=None)\n\
\n\
:Parameters:\n\
max_cache_entries : integer\n\
The maximum number of entries in the cache. If ZERO the server\n\
default value is used (10,000).\n\
\n\
ssl2_timeout : integer\n\
The lifetime in seconds of an SSL2 session. The minimum timeout\n\
value is 5 seconds and the maximum is 24 hours. Values outside\n\
this range are replaced by the server default value (100 seconds).\n\
\n\
ssl3_timeout : integer\n\
The lifetime in seconds of an SSL3 session. The minimum timeout\n\
value is 5 seconds and the maximum is 24 hours. Values outside\n\
this range are replaced by the server default value (24 hours).\n\
\n\
directory : string\n\
A string specifying the pathname of the directory that will\n\
contain the session cache. If None the server default value is\n\
used (/tmp (Unix) or \\temp (NT)).\n\
\n\
This function sets up a Server Session ID (SID) cache that is safe for\n\
access by multiple processes on the same system.\n\
\n\
Like `ssl.config_server_session_id_cache()`, with one important\n\
difference. If the application will run multiple processes (as\n\
opposed to, or in addition to multiple threads), then it must call\n\
this function, instead of calling\n\
`ssl.config_server_session_id_cache()`. This has nothing to do with\n\
the number of processors, only processes.\n\
");
static PyObject *
SSL_config_mp_server_sid_cache(PyObject *self, PyObject *args, PyObject *kwds)
{
static char *kwlist[] = {"max_cache_entries", "ssl2_timeout", "ssl3_timeout", "directory", NULL};
int max_cache_entries = 0;
PRUint32 ssl2_timeout = 0;
PRUint32 ssl3_timeout = 0;
PyObject *py_directory = Py_None;
PyObject *py_directory_utf8 = NULL;
char *directory = NULL;
TraceMethodEnter(self);
if (!PyArg_ParseTupleAndKeywords(args, kwds, "|iIIO:config_mp_server_sid_cache", kwlist,
&max_cache_entries, &ssl2_timeout, &ssl3_timeout, &py_directory))
return NULL;
if (PyString_Check(py_directory) || PyUnicode_Check(py_directory)) {
if (PyString_Check(py_directory)) {
py_directory_utf8 = py_directory;
Py_INCREF(py_directory_utf8);
} else {
py_directory_utf8 = PyUnicode_AsUTF8String(py_directory);
}
directory = PyString_AsString(py_directory_utf8);
} else if (PyNone_Check(py_directory)) {
directory = NULL;
} else {
PyErr_Format(PyExc_TypeError, "directory must be string or None, not %.200s",
Py_TYPE(py_directory)->tp_name);
return NULL;
}
if (SSL_ConfigMPServerSIDCache(max_cache_entries, ssl2_timeout,
ssl3_timeout, directory) != SECSuccess) {
Py_XDECREF(py_directory_utf8);
return set_nspr_error(NULL);
}
Py_XDECREF(py_directory_utf8);
Py_RETURN_NONE;
}
PyDoc_STRVAR(SSL_get_max_server_cache_locks_doc,
"get_max_server_cache_locks() -> int\n\
\n\
Get the configured maximum number of mutexes used for the server's\n\
store of SSL sessions. This value is used by the server session ID\n\
cache initialization functions.\n\
");
static PyObject *
SSL_get_max_server_cache_locks(PyObject *self, PyObject *args)
{
TraceMethodEnter(self);
return PyInt_FromLong(SSL_GetMaxServerCacheLocks());
}
PyDoc_STRVAR(SSL_set_max_server_cache_locks_doc,
"set_max_server_cache_locks(max_locks)\n\
\n\
:Parameters:\n\
max_locks : int\n\
Maximum number of locks\n\
\n\
Set the configured maximum number of mutexes used for the server's\n\
store of SSL sessions. This value is used by the server session ID\n\
cache initialization functions. Note that on some platforms, these\n\
mutexes are actually implemented with POSIX semaphores, or with\n\
unnamed pipes. The default value varies by platform. An attempt to\n\
set a too-low maximum will return an error and the configured value\n\
will not be changed.\n\
");
static PyObject *
SSL_set_max_server_cache_locks(PyObject *self, PyObject *args)
{
unsigned int max_locks;
TraceMethodEnter(self);
if (!PyArg_ParseTuple(args, "I:set_max_server_cache_locks",
&max_locks))
return NULL;
if (SSL_SetMaxServerCacheLocks(max_locks) != SECSuccess) {
return set_nspr_error(NULL);
}
Py_RETURN_NONE;
}
PyDoc_STRVAR(SSL_clear_session_cache_doc,
@@ -2259,10 +2551,26 @@ in RAM (not on disk).\n\
static PyObject *
SSL_clear_session_cache(PyObject *self, PyObject *args)
{
TraceMethodEnter(self);
SSL_ClearSessionCache();
Py_RETURN_NONE;
}
PyDoc_STRVAR(SSL_shutdown_server_session_id_cache_doc,
"shutdown_server_session_id_cache()\n\
\n\
");
static PyObject *
SSL_shutdown_server_session_id_cache(PyObject *self, PyObject *args)
{
TraceMethodEnter(self);
SSL_ShutdownServerSessionIDCache();
Py_RETURN_NONE;
}
PyDoc_STRVAR(NSS_set_domestic_policy_doc,
"set_domestic_policy()\n\
\n\
@@ -2274,6 +2582,8 @@ features.\n\
static PyObject *
NSS_set_domestic_policy(PyObject *self, PyObject *args)
{
TraceMethodEnter(self);
if (NSS_SetDomesticPolicy() != SECSuccess) {
return set_nspr_error(NULL);
}
@@ -2292,6 +2602,8 @@ features.\n\
static PyObject *
NSS_set_export_policy(PyObject *self, PyObject *args)
{
TraceMethodEnter(self);
if (NSS_SetExportPolicy() != SECSuccess) {
return set_nspr_error(NULL);
}
@@ -2309,6 +2621,8 @@ regulations related to software products with encryption features.\n\
static PyObject *
NSS_set_france_policy(PyObject *self, PyObject *args)
{
TraceMethodEnter(self);
if (NSS_SetFrancePolicy() != SECSuccess) {
return set_nspr_error(NULL);
}
@@ -2318,20 +2632,25 @@ NSS_set_france_policy(PyObject *self, PyObject *args)
/* List of functions exported by this module. */
static PyMethodDef module_methods[] = {
{"nssinit", (PyCFunction)NSSinit, METH_VARARGS, NSSinit_doc},
{"nss_init", (PyCFunction)NSS_init, METH_VARARGS, NSS_init_doc},
{"nss_shutdown", (PyCFunction)NSS_shutdown, METH_NOARGS, NSS_shutdown_doc},
{"set_ssl_default_option", (PyCFunction)SSL_set_ssl_default_option, METH_VARARGS, SSL_set_ssl_default_option_doc},
{"get_ssl_default_option", (PyCFunction)SSL_get_ssl_default_option, METH_VARARGS, SSL_get_ssl_default_option_doc},
{"set_default_cipher_pref", (PyCFunction)SSL_set_default_cipher_pref, METH_VARARGS, SSL_set_default_cipher_pref_doc},
{"get_default_cipher_pref", (PyCFunction)SSL_get_default_cipher_pref, METH_VARARGS, SSL_get_default_cipher_pref_doc},
{"set_cipher_policy", (PyCFunction)SSL_set_cipher_policy, METH_VARARGS, SSL_set_cipher_policy_doc},
{"get_cipher_policy", (PyCFunction)SSL_get_cipher_policy, METH_VARARGS, SSL_get_cipher_policy_doc},
{"config_server_session_id_cache", (PyCFunction)SSL_config_server_session_id_cache, METH_VARARGS|METH_KEYWORDS, SSL_config_server_session_id_cache_doc},
{"clear_session_cache", (PyCFunction)SSL_clear_session_cache, METH_NOARGS, SSL_clear_session_cache_doc},
{"set_domestic_policy", (PyCFunction)NSS_set_domestic_policy, METH_NOARGS, NSS_set_domestic_policy_doc},
{"set_export_policy", (PyCFunction)NSS_set_export_policy, METH_NOARGS, NSS_set_export_policy_doc},
{"set_france_policy", (PyCFunction)NSS_set_france_policy, METH_NOARGS, NSS_set_france_policy_doc},
{"nssinit", (PyCFunction)NSSinit, METH_VARARGS, NSSinit_doc},
{"nss_init", (PyCFunction)NSS_init, METH_VARARGS, NSS_init_doc},
{"nss_shutdown", (PyCFunction)NSS_shutdown, METH_NOARGS, NSS_shutdown_doc},
{"set_ssl_default_option", (PyCFunction)SSL_set_ssl_default_option, METH_VARARGS, SSL_set_ssl_default_option_doc},
{"get_ssl_default_option", (PyCFunction)SSL_get_ssl_default_option, METH_VARARGS, SSL_get_ssl_default_option_doc},
{"set_default_cipher_pref", (PyCFunction)SSL_set_default_cipher_pref, METH_VARARGS, SSL_set_default_cipher_pref_doc},
{"get_default_cipher_pref", (PyCFunction)SSL_get_default_cipher_pref, METH_VARARGS, SSL_get_default_cipher_pref_doc},
{"set_cipher_policy", (PyCFunction)SSL_set_cipher_policy, METH_VARARGS, SSL_set_cipher_policy_doc},
{"get_cipher_policy", (PyCFunction)SSL_get_cipher_policy, METH_VARARGS, SSL_get_cipher_policy_doc},
{"config_server_session_id_cache", (PyCFunction)SSL_config_server_session_id_cache, METH_VARARGS|METH_KEYWORDS, SSL_config_server_session_id_cache_doc},
{"config_mp_server_sid_cache", (PyCFunction)SSL_config_mp_server_sid_cache, METH_VARARGS|METH_KEYWORDS, SSL_config_mp_server_sid_cache_doc},
{"config_server_session_id_cache_with_opt", (PyCFunction)SSL_config_server_session_id_cache_with_opt, METH_VARARGS|METH_KEYWORDS, SSL_config_server_session_id_cache_with_opt_doc},
{"get_max_server_cache_locks", (PyCFunction)SSL_get_max_server_cache_locks, METH_NOARGS, SSL_get_max_server_cache_locks_doc},
{"set_max_server_cache_locks", (PyCFunction)SSL_set_max_server_cache_locks, METH_VARARGS, SSL_set_max_server_cache_locks_doc},
{"clear_session_cache", (PyCFunction)SSL_clear_session_cache, METH_NOARGS, SSL_clear_session_cache_doc},
{"shutdown_server_session_id_cache", (PyCFunction)SSL_shutdown_server_session_id_cache, METH_NOARGS, SSL_shutdown_server_session_id_cache_doc},
{"set_domestic_policy", (PyCFunction)NSS_set_domestic_policy, METH_NOARGS, NSS_set_domestic_policy_doc},
{"set_export_policy", (PyCFunction)NSS_set_export_policy, METH_NOARGS, NSS_set_export_policy_doc},
{"set_france_policy", (PyCFunction)NSS_set_france_policy, METH_NOARGS, NSS_set_france_policy_doc},
{NULL, NULL} /* Sentinel */
};

241
mozilla/security/python/nss/test/cipher_test.py
View File

@@ -1,241 +0,0 @@
#!/usr/bin/python
import sys
import os
import getopt
import nss.nss as nss
verbose = 0
def setup_contexts(mechanism, key, iv):
# Get a PK11 slot based on the cipher
slot = nss.get_best_slot(mechanism)
# If key was supplied use it, otherwise generate one
if key:
if verbose:
print "using supplied key data"
print "key:\n%s" % (key)
key_si = nss.SecItem(nss.read_hex(key))
sym_key = nss.import_sym_key(slot, mechanism, nss.PK11_OriginUnwrap,
nss.CKA_ENCRYPT, key_si)
else:
if verbose:
print "generating key data"
sym_key = slot.key_gen(mechanism, None, slot.get_best_key_length(mechanism))
# If initialization vector was supplied use it, otherwise set it to None
if iv:
if verbose:
print "supplied iv:\n%s" % (iv)
iv_data = nss.read_hex(iv)
iv_si = nss.SecItem(iv_data)
iv_param = nss.param_from_iv(mechanism, iv_si)
else:
iv_length = nss.get_iv_length(mechanism)
if iv_length > 0:
iv_data = nss.generate_random(iv_length)
iv_si = nss.SecItem(iv_data)
iv_param = nss.param_from_iv(mechanism, iv_si)
if verbose:
print "generated %d byte initialization vector: %s" % \
(iv_length, nss.data_to_hex(iv_data, separator=":"))
else:
iv_param = None
# Create an encoding context
encoding_ctx = nss.create_context_by_sym_key(mechanism, nss.CKA_ENCRYPT,
sym_key, iv_param)
# Create a decoding context
decoding_ctx = nss.create_context_by_sym_key(mechanism, nss.CKA_DECRYPT,
sym_key, iv_param)
return encoding_ctx, decoding_ctx
def simple_test(encoding_ctx, decoding_ctx, plain_text):
result = 0
if verbose:
print "Plain Text:\n%s" % (plain_text)
# Encode the plain text by feeding it to cipher_op getting cipher text back.
# Append the final bit of cipher text by calling digest_final
cipher_text = encoding_ctx.cipher_op(plain_text)
cipher_text += encoding_ctx.digest_final()
if verbose:
print "Cipher Text:\n%s" % (nss.data_to_hex(cipher_text, separator=":"))
# Decode the cipher text by feeding it to cipher_op getting plain text back.
# Append the final bit of plain text by calling digest_final
decoded_text = decoding_ctx.cipher_op(cipher_text)
decoded_text += decoding_ctx.digest_final()
if verbose:
print "Decoded Text:\n%s" % (decoded_text)
# Validate the encryption/decryption by comparing the decoded text with
# the original plain text, they should match.
if decoded_text != plain_text:
result = 1
print "FAILED! decoded_text != plain_text"
if cipher_text == plain_text:
result = 1
print "FAILED! cipher_text == plain_text"
return result
def file_test(encoding_ctx, decoding_ctx, in_filename, chunk_size):
result = 0
encrypted_filename = os.path.basename(in_filename) + ".encrypted"
decrypted_filename = os.path.basename(in_filename) + ".decrypted"
in_file = open(in_filename, "r")
encrypted_file = open(encrypted_filename, "w")
if verbose:
print "Encrypting file \"%s\" to \"%s\"" % (in_filename, encrypted_filename)
# Encode the data read from a file in chunks
while True:
# Read a chunk of data until EOF, encrypt it and write the encrypted data
in_data = in_file.read(chunk_size)
if len(in_data) == 0: # EOF
break
encrypted_data = encoding_ctx.cipher_op(in_data)
encrypted_file.write(encrypted_data)
# Done encoding the input, get the final encoded data, write it, close files
encrypted_data = encoding_ctx.digest_final()
encrypted_file.write(encrypted_data)
in_file.close()
encrypted_file.close()
# Decode the encoded file in a similar fashion
if verbose:
print "Decrypting file \"%s\" to \"%s\"" % (encrypted_filename, decrypted_filename)
encrypted_file = open(encrypted_filename, "r")
decrypted_file = open(decrypted_filename, "w")
while True:
# Read a chunk of data until EOF, encrypt it and write the encrypted data
in_data = encrypted_file.read(chunk_size)
if len(in_data) == 0: # EOF
break
decrypted_data = decoding_ctx.cipher_op(in_data)
decrypted_file.write(decrypted_data)
# Done encoding the input, get the final encoded data, write it, close files
decrypted_data = decoding_ctx.digest_final()
decrypted_file.write(decrypted_data)
encrypted_file.close()
decrypted_file.close()
# Validate the encryption/decryption by comparing the decoded text with
# the original plain text, they should match.
in_data = open(in_filename).read()
encrypted_data = open(encrypted_filename).read()
decrypted_data = open(decrypted_filename).read()
if decrypted_data != in_data:
result = 1
print "FAILED! decrypted_data != in_data"
if encrypted_data == in_data:
result = 1
print "FAILED! encrypted_data == in_data"
# clean up
os.unlink(encrypted_filename)
os.unlink(decrypted_filename)
return result
def usage():
print '''\
digest_test [-v -h] filename
filename file to be used as test data
-v --verbose turn on verbose output
-h --help print usage
-s --size number of octets processed in one iteration
-m --mech encryption mechanism name (e.g. CKM_*)
name is case insensitive, CKM_ prefix is optional
-t --text plain text
-k --key key (in hexadecimal format)
-i --iv initialization vector (in hexadecimal format)
'''
def main():
global verbose
mechanism = nss.CKM_DES_CBC_PAD
plain_text = "Encrypt me!"
key = "e8:a7:7c:e2:05:63:6a:31"
iv = "e4:bb:3b:d3:c3:71:2e:58"
in_filename = None
chunk_size = 128
try:
opts, args = getopt.getopt(sys.argv[1:], "hvs:m:t:k:i:",
["help", "verbose", "size=", "mechanism=", "text=",
"key=", "iv="])
except getopt.GetoptError, err:
print str(err) # will print something like "option -a not recognized"
usage()
sys.exit(2)
verbose = False
for o, a in opts:
if o in ("-s", "--size"):
chunk_size = int(a)
elif o in ("-m", "--mech"):
try:
mechanism = nss.key_mechanism_type_from_name(a)
except Exception, e:
print "error with mech argument (%s)" % (e)
sys.exit(2)
elif o in ("-t", "--text"):
plain_text = a
elif o in ("-k", "--key"):
key = a
elif o in ("-i", "--iv"):
iv = a
elif o in ("-v", "--verbose"):
verbose += 1
elif o in ("-h", "--help"):
usage()
sys.exit(0)
else:
assert False, "unhandled option"
if (len(args) > 1):
print "expected single file name"
usage()
sys.exit(2)
elif (len(args) == 1):
in_filename = args[0]
nss.nss_init_nodb()
result = 0
encoding_ctx, decoding_ctx = setup_contexts(mechanism, key, iv)
result += simple_test(encoding_ctx, decoding_ctx, plain_text)
if in_filename:
# In theory we should be able to reuse a context by calling finalize()
# on it, however at the time of this writing it only works for
# digest contexts, not encryption/decryption contexts
# so as a workaround we just create the contexts again
#
#encoding_ctx.finalize()
#decoding_ctx.finalize()
encoding_ctx, decoding_ctx = setup_contexts(mechanism, key, iv)
result += file_test(encoding_ctx, decoding_ctx, in_filename, chunk_size)
if result == 0:
print "SUCCESS"
else:
print "FAILED %d tests" % (result)
sys.exit(result)
if __name__ == "__main__":
main()

164
mozilla/security/python/nss/test/digest_test.py
View File

@@ -1,164 +0,0 @@
#!/usr/bin/python
import subprocess
import sys
import os
import getopt
import nss.nss as nss
verbose = False
tests = [
{'name' : 'md5',
'ref_cmd' : 'md5sum',
'nss_digest_func' : nss.md5_digest,
'hash_oid' : nss.SEC_OID_MD5},
{'name' : 'sha1',
'ref_cmd' : 'sha1sum',
'nss_digest_func' : nss.sha1_digest,
'hash_oid' : nss.SEC_OID_SHA1},
{'name' : 'sha256',
'ref_cmd' : 'sha256sum',
'nss_digest_func' : nss.sha256_digest,
'hash_oid' : nss.SEC_OID_SHA256},
{'name' : 'sha512',
'ref_cmd' : 'sha512sum',
'nss_digest_func' : nss.sha512_digest,
'hash_oid' : nss.SEC_OID_SHA512},
]
def do_test(name, ref_cmd, nss_digest_func, hash_oid, in_filename, chunk_size):
result = 0
hash_oid_name = nss.oid_str(hash_oid)
if verbose:
print 'running test %s: nss_digest_func=%s hash_oid=%s in_filename=%s' % \
(name, nss_digest_func.__name__, hash_oid_name, in_filename)
# read the data in from the file
ref_data = open(in_filename).read()
# Run the system hash function to get a reference result.
# Since we're testing the python-nss binding we assume
# the system command is entirely independent and correct.
#
# Because our digest routines return raw data (e.g. a buffer of octets)
# and the system hash command returns a hex string which we need to compare agains,
# and because we sometimes want to print the result of our digest functions
# always convert our results to a hex string via nss.data_to_hex()
proc = subprocess.Popen([ref_cmd, in_filename], stdout=subprocess.PIPE)
status = proc.wait();
reference_digest = proc.stdout.read().split()[0]
if verbose:
print 'reference_digest\n%s' % (reference_digest)
# Run the test with convenience digest function (e.g. nss.sha256_digest, etc.).
test_digest = nss.data_to_hex(nss_digest_func(ref_data), separator=None)
if verbose: print 'nss %s\n%s' % (nss_digest_func.__name__, test_digest)
if test_digest != reference_digest:
result += 1
print 'nss %s test failed' % (nss_digest_func.__name__)
print 'reference = %s' % (reference_digest)
print 'test = %s' % (test_digest)
# Run the test using the generic hash_buf function specifying the hash algorithm.
test_digest = nss.data_to_hex(nss.hash_buf(hash_oid, ref_data), separator=None)
if verbose: print 'nss.hash_buf %s\n%s' % (hash_oid_name, test_digest)
if test_digest != reference_digest:
result += 1
print 'nss.hash_buf %s test failed' % (hash_oid_name)
print 'reference = %s' % (reference_digest)
print 'test = %s' % (test_digest)
# Run the test using the lowest level hashing functions by specifying the hash algorithm.
# The entire input data is supplied all at once in a single call.
context = nss.create_digest_context(hash_oid)
context.digest_begin()
context.digest_op(ref_data)
test_digest = nss.data_to_hex(context.digest_final(), separator=None)
if verbose: print 'nss.digest_context %s\n%s' % (hash_oid_name, test_digest)
if test_digest != reference_digest:
result += 1
print 'nss.digest_context %s test failed' % (hash_oid_name)
print 'reference = %s' % (reference_digest)
print 'test = %s' % (test_digest)
# Run the test using the lowest level hashing functions by specifying the hash algorithm
# and feeding 'chunks' of data one at a time to be consumed.
in_file = open(in_filename, 'r')
context = nss.create_digest_context(hash_oid)
context.digest_begin()
while True:
in_data = in_file.read(chunk_size)
if len(in_data) == 0:
break
context.digest_op(in_data)
test_digest = nss.data_to_hex(context.digest_final(), separator=None)
if verbose: print 'chunked nss.digest_context %s\n%s' % (hash_oid_name, test_digest)
if test_digest != reference_digest:
result += 1
print 'chunked nss.digest_context %s test failed' % (hash_oid_name)
print 'reference = %s' % (reference_digest)
print 'test = %s' % (test_digest)
return result
def usage():
print '''\
digest_test [-v -h] filename
-v --verbose turn on verbose output
-h --help print usage
-s --size number of octets processed in one iteration
-i --infile filename file to be used as test data
'''
def main():
global verbose
in_filename = sys.argv[0]
chunk_size = 128
try:
opts, args = getopt.getopt(sys.argv[1:], 'hvs:i:',
['help', 'verbose', 'size=', 'infile='])
except getopt.GetoptError, err:
print str(err) # will print something like 'option -a not recognized'
usage()
sys.exit(2)
verbose = False
for o, a in opts:
if o in ('-s', '--size'):
chunk_size = int(a)
elif o in ('-i', '--infile'):
in_filename = a
elif o in ('-v', '--verbose'):
verbose = True
elif o in ('-h', '--help'):
usage()
sys.exit(0)
else:
assert False, 'unhandled option'
nss.nss_init_nodb()
result = 0
for test in tests:
result += do_test(test['name'], test['ref_cmd'], test['nss_digest_func'],
test['hash_oid'], in_filename, chunk_size)
if result == 0:
print 'SUCCESS'
else:
print 'FAILED %d tests' % (result)
sys.exit(result)
if __name__ == '__main__':
main()

124
mozilla/security/python/nss/test/run_tests Executable file
View File

@@ -0,0 +1,124 @@
#!/usr/bin/python
import getopt
import sys
import os
import unittest
from util import get_build_dir
#-------------------------------------------------------------------------------
prog_name = os.path.basename(sys.argv[0])
config = {
'in_tree' : True,
}
#-------------------------------------------------------------------------------
def run_tests():
import setup_certs
import test_cert_components
import test_cipher
import test_digest
import test_pkcs12
import test_client_server
setup_certs.setup_certs()
loader = unittest.TestLoader()
runner = unittest.TextTestRunner()
suite = loader.loadTestsFromModule(test_cert_components)
suite.addTests(loader.loadTestsFromModule(test_cipher))
suite.addTests(loader.loadTestsFromModule(test_digest))
suite.addTests(loader.loadTestsFromModule(test_pkcs12))
suite.addTests(loader.loadTestsFromModule(test_client_server))
result = runner.run(suite)
return len(result.failures)
#-------------------------------------------------------------------------------
class Usage(Exception):
def __init__(self, msg):
self.msg = msg
def usage():
'Print command help.'
return '''\
%(prog_name)s [-i]
-h --help print help
-i --installed runs the test using installed libraries
instead of "in tree" libraries
Runs unit tests.
By default test is done "in tree".
Examples:
Run test using libraries built in this tree
%(prog_name)s
Run post install test
%(prog_name)s -i
''' % {'prog_name' : prog_name,
}
#-------------------------------------------------------------------------------
def main(argv=None):
if argv is None:
argv = sys.argv
try:
try:
opts, args = getopt.getopt(argv[1:], 'hi',
['help', 'installed',])
except getopt.GetoptError, e:
raise Usage(e)
return 2
for o, a in opts:
if o in ('-h', '--help'):
print >>sys.stdout, usage()
return 0
elif o in ('-i', '--installed'):
config['in_tree'] = False
else:
raise Usage("command argument '%s' not handled, internal error" % o)
except Usage, e:
print >>sys.stderr, e.msg
print >>sys.stderr, "for help use --help"
return 2
if config['in_tree']:
# Run the tests "in the tree"
# Rather than testing with installed versions run the test
# with the package built in this tree.
build_dir = get_build_dir()
if build_dir and os.path.exists(build_dir):
print "Using local libraries from tree, located here:\n%s\n" % build_dir
sys.path.insert(0, build_dir)
else:
print >>sys.stderr, "ERROR: Unable to locate in tree libraries"
return 2
else:
print "Using installed libraries"
num_failures = run_tests()
if num_failures == 0:
return 0
else:
return 1
#-------------------------------------------------------------------------------
if __name__ == '__main__':
sys.exit(main())

10
mozilla/security/python/nss/test/setup_certs.py
View File

@@ -28,11 +28,11 @@ client_username = 'test_user'
config = {
'verbose' : False,
'debug' : False,
'logfile' : prog_name + '.log',
'logfile' : None,
'log_level' : logging.INFO,
'interactive' : sys.stdout.isatty(),
'dbdir' : 'pki',
'db_passwd' : 'passwd',
'dbdir' : os.path.join(os.path.dirname(sys.argv[0]), 'pki'),
'db_passwd' : 'db_passwd',
'noise_file' : 'noise_file',
'ca_subject' : 'CN=Test CA',
'ca_nickname' : 'test_ca',
@@ -134,6 +134,8 @@ def run_cmd_with_password(cmd, password_prompt, password):
#-------------------------------------------------------------------------------
def setup_certs():
print "setting up certs ..."
if os.path.exists(config['dbdir']):
shutil.rmtree(config['dbdir'])
os.makedirs(config['dbdir'])
@@ -186,7 +188,7 @@ def setup_certs():
# 5. Import public root CA's
cmd = 'modutil -dbdir %s -add ca_certs -libfile /usr/lib/libnssckbi.so' % \
cmd = 'modutil -dbdir %s -add ca_certs -libfile libnssckbi.so' % \
(config['dbdir'])
exit_code, stdout, stderr = run_cmd(cmd)

170
mozilla/security/python/nss/test/test_cipher.py Executable file
View File

@@ -0,0 +1,170 @@
#!/usr/bin/python
import sys
import os
import unittest
import nss.nss as nss
#-------------------------------------------------------------------------------
verbose = False
mechanism = nss.CKM_DES_CBC_PAD
plain_text = "Encrypt me!"
key = "e8:a7:7c:e2:05:63:6a:31"
iv = "e4:bb:3b:d3:c3:71:2e:58"
in_filename = sys.argv[0]
chunk_size = 128
#-------------------------------------------------------------------------------
def setup_contexts(mechanism, key, iv):
# Get a PK11 slot based on the cipher
slot = nss.get_best_slot(mechanism)
# If key was supplied use it, otherwise generate one
if key:
if verbose:
print "using supplied key data"
print "key:\n%s" % (key)
key_si = nss.SecItem(nss.read_hex(key))
sym_key = nss.import_sym_key(slot, mechanism, nss.PK11_OriginUnwrap,
nss.CKA_ENCRYPT, key_si)
else:
if verbose:
print "generating key data"
sym_key = slot.key_gen(mechanism, None, slot.get_best_key_length(mechanism))
# If initialization vector was supplied use it, otherwise set it to None
if iv:
if verbose:
print "supplied iv:\n%s" % (iv)
iv_data = nss.read_hex(iv)
iv_si = nss.SecItem(iv_data)
iv_param = nss.param_from_iv(mechanism, iv_si)
else:
iv_length = nss.get_iv_length(mechanism)
if iv_length > 0:
iv_data = nss.generate_random(iv_length)
iv_si = nss.SecItem(iv_data)
iv_param = nss.param_from_iv(mechanism, iv_si)
if verbose:
print "generated %d byte initialization vector: %s" % \
(iv_length, nss.data_to_hex(iv_data, separator=":"))
else:
iv_param = None
# Create an encoding context
encoding_ctx = nss.create_context_by_sym_key(mechanism, nss.CKA_ENCRYPT,
sym_key, iv_param)
# Create a decoding context
decoding_ctx = nss.create_context_by_sym_key(mechanism, nss.CKA_DECRYPT,
sym_key, iv_param)
return encoding_ctx, decoding_ctx
#-------------------------------------------------------------------------------
class TestCipher(unittest.TestCase):
def setUp(self):
nss.nss_init_nodb()
self.encoding_ctx, self.decoding_ctx = setup_contexts(mechanism, key, iv)
def tearDown(self):
del self.encoding_ctx
del self.decoding_ctx
nss.nss_shutdown()
def test_string(self):
if verbose:
print "Plain Text:\n%s" % (plain_text)
# Encode the plain text by feeding it to cipher_op getting cipher text back.
# Append the final bit of cipher text by calling digest_final
cipher_text = self.encoding_ctx.cipher_op(plain_text)
cipher_text += self.encoding_ctx.digest_final()
if verbose:
print "Cipher Text:\n%s" % (nss.data_to_hex(cipher_text, separator=":"))
# Decode the cipher text by feeding it to cipher_op getting plain text back.
# Append the final bit of plain text by calling digest_final
decoded_text = self.decoding_ctx.cipher_op(cipher_text)
decoded_text += self.decoding_ctx.digest_final()
if verbose:
print "Decoded Text:\n%s" % (decoded_text)
# Validate the encryption/decryption by comparing the decoded text with
# the original plain text, they should match.
self.assertEqual(decoded_text, plain_text)
self.assertNotEqual(cipher_text, plain_text)
def test_file(self):
encrypted_filename = os.path.basename(in_filename) + ".encrypted"
decrypted_filename = os.path.basename(in_filename) + ".decrypted"
in_file = open(in_filename, "r")
encrypted_file = open(encrypted_filename, "w")
if verbose:
print "Encrypting file \"%s\" to \"%s\"" % (in_filename, encrypted_filename)
# Encode the data read from a file in chunks
while True:
# Read a chunk of data until EOF, encrypt it and write the encrypted data
in_data = in_file.read(chunk_size)
if len(in_data) == 0: # EOF
break
encrypted_data = self.encoding_ctx.cipher_op(in_data)
encrypted_file.write(encrypted_data)
# Done encoding the input, get the final encoded data, write it, close files
encrypted_data = self.encoding_ctx.digest_final()
encrypted_file.write(encrypted_data)
in_file.close()
encrypted_file.close()
# Decode the encoded file in a similar fashion
if verbose:
print "Decrypting file \"%s\" to \"%s\"" % (encrypted_filename, decrypted_filename)
encrypted_file = open(encrypted_filename, "r")
decrypted_file = open(decrypted_filename, "w")
while True:
# Read a chunk of data until EOF, encrypt it and write the encrypted data
in_data = encrypted_file.read(chunk_size)
if len(in_data) == 0: # EOF
break
decrypted_data = self.decoding_ctx.cipher_op(in_data)
decrypted_file.write(decrypted_data)
# Done encoding the input, get the final encoded data, write it, close files
decrypted_data = self.decoding_ctx.digest_final()
decrypted_file.write(decrypted_data)
encrypted_file.close()
decrypted_file.close()
# Validate the encryption/decryption by comparing the decoded text with
# the original plain text, they should match.
in_data = open(in_filename).read()
encrypted_data = open(encrypted_filename).read()
decrypted_data = open(decrypted_filename).read()
if decrypted_data != in_data:
result = 1
print "FAILED! decrypted_data != in_data"
if encrypted_data == in_data:
result = 1
print "FAILED! encrypted_data == in_data"
# clean up
os.unlink(encrypted_filename)
os.unlink(decrypted_filename)
#-------------------------------------------------------------------------------
if __name__ == '__main__':
unittest.main()

358
mozilla/security/python/nss/test/test_client_server.py Executable file
View File

@@ -0,0 +1,358 @@
#!/usr/bin/python
import os
import sys
import errno
import signal
import time
import unittest
from nss.error import NSPRError
import nss.io as io
import nss.nss as nss
import nss.ssl as ssl
# -----------------------------------------------------------------------------
NO_CLIENT_CERT = 0
REQUEST_CLIENT_CERT_ONCE = 1
REQUIRE_CLIENT_CERT_ONCE = 2
REQUEST_CLIENT_CERT_ALWAYS = 3
REQUIRE_CLIENT_CERT_ALWAYS = 4
verbose = False
info = False
password = 'db_passwd'
use_ssl = True
client_cert_action = NO_CLIENT_CERT
certdir = os.path.join(os.path.dirname(sys.argv[0]), 'pki')
hostname = os.uname()[1]
server_nickname = 'test_server'
client_nickname = 'test_user'
port = 1234
timeout_secs = 3
family = io.PR_AF_INET
sleep_time = 1
# -----------------------------------------------------------------------------
# Callback Functions
# -----------------------------------------------------------------------------
def password_callback(slot, retry, password):
if password: return password
return getpass.getpass("Enter password: ");
def handshake_callback(sock):
if verbose: print "handshake complete, peer = %s" % (sock.get_peer_name())
def auth_certificate_callback(sock, check_sig, is_server, certdb):
if verbose: print "auth_certificate_callback: check_sig=%s is_server=%s" % (check_sig, is_server)
cert_is_valid = False
cert = sock.get_peer_certificate()
pin_args = sock.get_pkcs11_pin_arg()
if pin_args is None:
pin_args = ()
#if verbose: print "cert:\n%s" % cert
# Define how the cert is being used based upon the is_server flag. This may
# seem backwards, but isn't. If we're a server we're trying to validate a
# client cert. If we're a client we're trying to validate a server cert.
if is_server:
intended_usage = nss.certificateUsageSSLClient
else:
intended_usage = nss.certificateUsageSSLServer
try:
# If the cert fails validation it will raise an exception, the errno attribute
# will be set to the error code matching the reason why the validation failed
# and the strerror attribute will contain a string describing the reason.
approved_usage = cert.verify_now(certdb, check_sig, intended_usage, *pin_args)
except Exception, e:
print >>sys.stderr, "auth_certificate_callback: %s" % e
cert_is_valid = False
if verbose: print "Returning cert_is_valid = %s" % cert_is_valid
return cert_is_valid
if verbose: print "approved_usage = %s" % ', '.join(nss.cert_usage_flags(approved_usage))
# Is the intended usage a proper subset of the approved usage
if approved_usage & intended_usage:
cert_is_valid = True
else:
cert_is_valid = False
# If this is a server, we're finished
if is_server or not cert_is_valid:
if verbose: print "Returning cert_is_valid = %s" % cert_is_valid
return cert_is_valid
# Certificate is OK. Since this is the client side of an SSL
# connection, we need to verify that the name field in the cert
# matches the desired hostname. This is our defense against
# man-in-the-middle attacks.
hostname = sock.get_hostname()
if verbose: print "verifying socket hostname (%s) matches cert subject (%s)" % (hostname, cert.subject)
try:
# If the cert fails validation it will raise an exception
cert_is_valid = cert.verify_hostname(hostname)
except Exception, e:
print >>sys.stderr, "auth_certificate_callback: %s" % e
cert_is_valid = False
if verbose: print "Returning cert_is_valid = %s" % cert_is_valid
return cert_is_valid
if verbose: print "Returning cert_is_valid = %s" % cert_is_valid
return cert_is_valid
def client_auth_data_callback(ca_names, chosen_nickname, password, certdb):
cert = None
if chosen_nickname:
try:
cert = nss.find_cert_from_nickname(chosen_nickname, password)
priv_key = nss.find_key_by_any_cert(cert, password)
if verbose: print "client cert:\n%s" % cert
return cert, priv_key
except NSPRError, e:
print >>sys.stderr, "client_auth_data_callback: %s" % e
return False
else:
nicknames = nss.get_cert_nicknames(certdb, cert.SEC_CERT_NICKNAMES_USER)
for nickname in nicknames:
try:
cert = nss.find_cert_from_nickname(nickname, password)
if verbose: print "client cert:\n%s" % cert
if cert.check_valid_times():
if cert.has_signer_in_ca_names(ca_names):
priv_key = nss.find_key_by_any_cert(cert, password)
return cert, priv_key
except NSPRError, e:
print >>sys.stderr, "client_auth_data_callback: %s" % e
return False
# -----------------------------------------------------------------------------
# Client Implementation
# -----------------------------------------------------------------------------
def client(request):
if use_ssl:
if info: print "client: using SSL"
ssl.set_domestic_policy()
valid_addr = False
# Get the IP Address of our server
try:
addr_info = io.AddrInfo(hostname)
except Exception, e:
print >>sys.stderr, "client: could not resolve host address \"%s\"" % hostname
return
for net_addr in addr_info:
if family != io.PR_AF_UNSPEC:
if net_addr.family != family: continue
net_addr.port = port
if use_ssl:
sock = ssl.SSLSocket(net_addr.family)
# Set client SSL socket options
sock.set_ssl_option(ssl.SSL_SECURITY, True)
sock.set_ssl_option(ssl.SSL_HANDSHAKE_AS_CLIENT, True)
sock.set_hostname(hostname)
# Provide a callback which notifies us when the SSL handshake is complete
sock.set_handshake_callback(handshake_callback)
# Provide a callback to supply our client certificate info
sock.set_client_auth_data_callback(client_auth_data_callback, client_nickname,
password, nss.get_default_certdb())
# Provide a callback to verify the servers certificate
sock.set_auth_certificate_callback(auth_certificate_callback,
nss.get_default_certdb())
else:
sock = io.Socket(net_addr.family)
try:
if verbose: print "client trying connection to: %s" % (net_addr)
sock.connect(net_addr, timeout=io.seconds_to_interval(timeout_secs))
if verbose: print "client connected to: %s" % (net_addr)
valid_addr = True
break
except Exception, e:
sock.close()
print >>sys.stderr, "client: connection to: %s failed (%s)" % (net_addr, e)
if not valid_addr:
print >>sys.stderr, "Could not establish valid address for \"%s\" in family %s" % \
(hostname, io.addr_family_name(family))
return
# Talk to the server
try:
if info: print "client: sending \"%s\"" % (request)
sock.send(request)
buf = sock.recv(1024)
if not buf:
print >>sys.stderr, "client: lost connection"
sock.close()
return
if info: print "client: received \"%s\"" % (buf)
except Exception, e:
print >>sys.stderr, "client: %s" % e
try:
sock.close()
except:
pass
return
try:
sock.shutdown()
except Exception, e:
print >>sys.stderr, "client: %s" % e
try:
sock.close()
if use_ssl:
ssl.clear_session_cache()
except Exception, e:
print >>sys.stderr, "client: %s" % e
return buf
# -----------------------------------------------------------------------------
# Server Implementation
# -----------------------------------------------------------------------------
def server():
global family
if verbose: print "starting server:"
# Initialize
# Setup an IP Address to listen on any of our interfaces
if family == io.PR_AF_UNSPEC:
family = io.PR_AF_INET
net_addr = io.NetworkAddress(io.PR_IpAddrAny, port, family)
if use_ssl:
if info: print "server: using SSL"
ssl.set_domestic_policy()
nss.set_password_callback(password_callback)
# Perform basic SSL server configuration
ssl.set_default_cipher_pref(ssl.SSL_RSA_WITH_NULL_MD5, True)
ssl.config_server_session_id_cache()
# Get our certificate and private key
server_cert = nss.find_cert_from_nickname(server_nickname, password)
priv_key = nss.find_key_by_any_cert(server_cert, password)
server_cert_kea = server_cert.find_kea_type();
#if verbose: print "server cert:\n%s" % server_cert
sock = ssl.SSLSocket(net_addr.family)
# Set server SSL socket options
sock.set_pkcs11_pin_arg(password)
sock.set_ssl_option(ssl.SSL_SECURITY, True)
sock.set_ssl_option(ssl.SSL_HANDSHAKE_AS_SERVER, True)
# If we're doing client authentication then set it up
if client_cert_action >= REQUEST_CLIENT_CERT_ONCE:
sock.set_ssl_option(ssl.SSL_REQUEST_CERTIFICATE, True)
if client_cert_action == REQUIRE_CLIENT_CERT_ONCE:
sock.set_ssl_option(ssl.SSL_REQUIRE_CERTIFICATE, True)
sock.set_auth_certificate_callback(auth_certificate_callback, nss.get_default_certdb())
# Configure the server SSL socket
sock.config_secure_server(server_cert, priv_key, server_cert_kea)
else:
sock = io.Socket(net_addr.family)
# Bind to our network address and listen for clients
sock.bind(net_addr)
if verbose: print "listening on: %s" % (net_addr)
sock.listen()
while True:
# Accept a connection from a client
client_sock, client_addr = sock.accept()
if use_ssl:
client_sock.set_handshake_callback(handshake_callback)
if verbose: print "client connect from: %s" % (client_addr)
while True:
try:
# Handle the client connection
buf = client_sock.recv(1024)
if not buf:
print >>sys.stderr, "server: lost lost connection to %s" % (client_addr)
break
if info: print "server: received \"%s\"" % (buf)
reply = "{%s}" % buf # echo
if info: print "server: sending \"%s\"" % (reply)
client_sock.send(reply) # echo
time.sleep(sleep_time)
client_sock.shutdown()
client_sock.close()
break
except Exception, e:
print >>sys.stderr, "server: %s" % e
break
break
# Clean up
sock.shutdown()
sock.close()
if use_ssl:
ssl.shutdown_server_session_id_cache()
# -----------------------------------------------------------------------------
def run_server():
pid = os.fork()
if pid == 0:
nss.nss_init(certdir)
server()
nss.nss_shutdown()
time.sleep(sleep_time)
return pid
def cleanup_server(pid):
try:
wait_pid, wait_status = os.waitpid(pid, os.WNOHANG)
if wait_pid == 0:
os.kill(pid, signal.SIGKILL)
except OSError, e:
if e.errno == errno.ECHILD:
pass # child already exited
else:
print >>sys.stderr, "cleanup_server: %s" % e
class TestSSL(unittest.TestCase):
def setUp(self):
print
self.server_pid = run_server()
def tearDown(self):
cleanup_server(self.server_pid)
def test_ssl(self):
request = "foo"
nss.nss_init(certdir)
reply = client(request)
nss.nss_shutdown()
self.assertEqual("{%s}" % request, reply)
if __name__ == '__main__':
unittest.main()

111
mozilla/security/python/nss/test/test_digest.py Executable file
View File

@@ -0,0 +1,111 @@
#!/usr/bin/python
import subprocess
import sys
import unittest
import nss.nss as nss
#-------------------------------------------------------------------------------
verbose = False
in_filename = sys.argv[0]
chunk_size = 128
#-------------------------------------------------------------------------------
class TestDigest(unittest.TestCase):
def setUp(self):
nss.nss_init_nodb()
def tearDown(self):
nss.nss_shutdown()
def do_test(self, name, ref_cmd, nss_digest_func, hash_oid):
hash_oid_name = nss.oid_str(hash_oid)
if verbose:
print 'running test %s: nss_digest_func=%s hash_oid=%s in_filename=%s' % \
(name, nss_digest_func.__name__, hash_oid_name, in_filename)
# read the data in from the file
ref_data = open(in_filename).read()
# Run the system hash function to get a reference result.
# Since we're testing the python-nss binding we assume
# the system command is entirely independent and correct.
#
# Because our digest routines return raw data (e.g. a buffer of octets)
# and the system hash command returns a hex string which we need to compare agains,
# and because we sometimes want to print the result of our digest functions
# always convert our results to a hex string via nss.data_to_hex()
proc = subprocess.Popen([ref_cmd, in_filename], stdout=subprocess.PIPE)
status = proc.wait();
reference_digest = proc.stdout.read().split()[0]
if verbose:
print 'reference_digest\n%s' % (reference_digest)
# Run the test with convenience digest function (e.g. nss.sha256_digest, etc.).
test_digest = nss.data_to_hex(nss_digest_func(ref_data), separator=None)
if verbose: print 'nss %s\n%s' % (nss_digest_func.__name__, test_digest)
self.assertEqual(test_digest, reference_digest,
msg='nss %s test failed reference=%s test=%s' % \
(nss_digest_func.__name__, reference_digest, test_digest))
# Run the test using the generic hash_buf function specifying the hash algorithm.
test_digest = nss.data_to_hex(nss.hash_buf(hash_oid, ref_data), separator=None)
if verbose: print 'nss.hash_buf %s\n%s' % (hash_oid_name, test_digest)
self.assertEqual(test_digest, reference_digest,
msg='nss.hash_buf %s test failed reference=%s test=%s' % \
(hash_oid_name, reference_digest, test_digest))
# Run the test using the lowest level hashing functions by specifying the hash algorithm.
# The entire input data is supplied all at once in a single call.
context = nss.create_digest_context(hash_oid)
context.digest_begin()
context.digest_op(ref_data)
test_digest = nss.data_to_hex(context.digest_final(), separator=None)
if verbose: print 'nss.digest_context %s\n%s' % (hash_oid_name, test_digest)
self.assertEqual(test_digest, reference_digest,
msg='nss.digest_context %s test failed reference=%s test=%s' % \
(hash_oid_name, reference_digest, test_digest))
# Run the test using the lowest level hashing functions by specifying the hash algorithm
# and feeding 'chunks' of data one at a time to be consumed.
in_file = open(in_filename, 'r')
context = nss.create_digest_context(hash_oid)
context.digest_begin()
while True:
in_data = in_file.read(chunk_size)
if len(in_data) == 0:
break
context.digest_op(in_data)
test_digest = nss.data_to_hex(context.digest_final(), separator=None)
if verbose: print 'chunked nss.digest_context %s\n%s' % (hash_oid_name, test_digest)
self.assertEqual(test_digest, reference_digest,
msg='chunked nss.digest_context %s test failed reference=%s test=%s' % \
(hash_oid_name, reference_digest, test_digest))
def test_md5(self):
self.do_test('md5', 'md5sum', nss.md5_digest, nss.SEC_OID_MD5)
def test_sha1(self):
self.do_test('sha1', 'sha1sum', nss.sha1_digest, nss.SEC_OID_SHA1)
def test_sha256(self):
self.do_test('sha256', 'sha256sum', nss.sha256_digest, nss.SEC_OID_SHA256)
def test_sha512(self):
self.do_test('sha512', 'sha512sum', nss.sha512_digest, nss.SEC_OID_SHA512)
#-------------------------------------------------------------------------------
if __name__ == '__main__':
unittest.main()

134
mozilla/security/python/nss/test/test_pkcs12.py Executable file
View File

@@ -0,0 +1,134 @@
#!/usr/bin/python
import sys
import os
import subprocess
import shlex
import unittest
from nss.error import NSPRError
import nss.error as nss_error
import nss.nss as nss
#-------------------------------------------------------------------------------
verbose = False
certdir = 'pki'
db_passwd = 'db_passwd'
pkcs12_file_password = 'pk12_passwd'
read_nickname = 'test_user'
read_pkcs12_file = '%s.p12' % read_nickname
write_export_file = False
export_nickname = 'test_server'
#-------------------------------------------------------------------------------
def run_cmd(cmd):
if verbose: print "running command: %s" % cmd
args = shlex.split(cmd)
subprocess.check_call(args, stdout=subprocess.PIPE)
#-------------------------------------------------------------------------------
def password_callback(slot, retry):
return db_passwd
#-------------------------------------------------------------------------------
def nickname_collision_callback(old_nickname, cert):
cancel = False
new_nickname = cert.make_ca_nickname()
return new_nickname, cancel
#-------------------------------------------------------------------------------
def load_tests(loader, tests, pattern):
suite = unittest.TestSuite()
tests = loader.loadTestsFromNames(['test_pkcs12.TestPKCS12Decoder.test_read',
'test_pkcs12.TestPKCS12Decoder.test_import',
'test_pkcs12.TestPKCS12Export.test_export',
])
suite.addTests(tests)
return suite
#-------------------------------------------------------------------------------
class TestPKCS12Decoder(unittest.TestCase):
def setUp(self):
nss.nss_init_read_write(certdir)
nss.set_password_callback(password_callback)
nss.pkcs12_set_nickname_collision_callback(nickname_collision_callback)
nss.pkcs12_enable_all_ciphers()
def tearDown(self):
nss.nss_shutdown()
def test_read(self):
if verbose: print "test_read"
cmd='pk12util -o %s -n %s -d pki -K %s -W %s' % \
(read_pkcs12_file, read_nickname, db_passwd, pkcs12_file_password)
run_cmd(cmd)
slot = nss.get_internal_key_slot()
pkcs12 = nss.PKCS12Decoder(read_pkcs12_file, pkcs12_file_password, slot)
self.assertEqual(len(pkcs12), 3)
cert_bag_count = 0
key_seen = None
for bag in pkcs12:
if bag.type == nss.SEC_OID_PKCS12_V1_CERT_BAG_ID:
self.assertIsNone(bag.shroud_algorithm_id)
cert_bag_count += 1
if key_seen is None:
key_seen = bag.has_key
elif key_seen is True:
self.assertIs(bag.has_key, False)
elif key_seen is False:
self.assertIs(bag.has_key, True)
else:
self.fail("unexpected has_key for bag type = %s(%d)" % (bag.has_key, nss.oid_tag_name(bag.type), bag.type))
elif bag.type == nss.SEC_OID_PKCS12_V1_PKCS8_SHROUDED_KEY_BAG_ID:
self.assertIsInstance(bag.shroud_algorithm_id, nss.AlgorithmID)
self.assertIs(bag.has_key, False)
else:
self.fail("unexpected bag type = %s(%d)" % (nss.oid_tag_name(bag.type), bag.type))
self.assertEqual(cert_bag_count, 2)
def test_import(self):
if verbose: print "test_import"
cmd='certutil -d pki -D -n %s' % (read_nickname)
run_cmd(cmd)
slot = nss.get_internal_key_slot()
pkcs12 = nss.PKCS12Decoder(read_pkcs12_file, pkcs12_file_password, slot)
slot.authenticate()
pkcs12.database_import()
#-------------------------------------------------------------------------------
class TestPKCS12Export(unittest.TestCase):
def setUp(self):
nss.nss_init(certdir)
nss.set_password_callback(password_callback)
nss.pkcs12_enable_all_ciphers()
def tearDown(self):
nss.nss_shutdown()
def test_export(self):
if verbose: print "test_export"
pkcs12_data = nss.pkcs12_export(export_nickname, pkcs12_file_password)
if write_export_file:
p12_file_path = os.path.join(os.path.dirname(sys.argv[0]), "%s.p12" % export_nickname)
f = open(p12_file_path, 'w')
f.write(pkcs12_data)
f.close()
if __name__ == '__main__':
unittest.main()