Coverage for ipaserver/plugins/selfsign : 52%

# Authors: 

#   Rob Crittenden <rcritten@@redhat.com> 

#   John Dennis <jdennis@redhat.com> 

# 

# Copyright (C) 2009  Red Hat 

# see file 'COPYING' for use and warranty information 

# 

# This program is free software; you can redistribute it and/or modify 

# it under the terms of the GNU General Public License as published by 

# the Free Software Foundation, either version 3 of the License, or 

# (at your option) any later version. 

# 

# This program is distributed in the hope that it will be useful, 

# but WITHOUT ANY WARRANTY; without even the implied warranty of 

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the 

# GNU General Public License for more details. 

# 

# You should have received a copy of the GNU General Public License 

# along with this program.  If not, see <http://www.gnu.org/licenses/>. 

""" 

Backend plugin for RA activities. 

The `ra` plugin provides access to the CA to issue, retrieve, and revoke 

certificates via the following methods: 

    * `ra.check_request_status()` - check certificate request status. 

    * `ra.get_certificate()` - retrieve an existing certificate. 

    * `ra.request_certificate()` - request a new certificate. 

    * `ra.revoke_certificate()` - revoke a certificate. 

    * `ra.take_certificate_off_hold()` - take a certificate off hold. 

""" 

from ipalib import api, SkipPluginModule 

if api.env.ra_plugin != 'selfsign': 

    # In this case, abort loading this plugin module... 

    raise SkipPluginModule(reason='selfsign is not selected as RA plugin, it is %s' % api.env.ra_plugin) 

from ipalib import Backend 

from ipalib import errors 

from ipalib import x509 

from ipalib import pkcs10 

from ipapython.certdb import get_ca_nickname 

import subprocess 

import os 

import re 

from ipaserver.plugins import rabase 

from ipaserver.install import certs 

import tempfile 

from ipalib import  _ 

from ipalib import  api 

from ipalib.plugins.cert import get_csr_hostname 

from nss.error import NSPRError 

class ra(rabase.rabase): 

    """ 

    Request Authority backend plugin. 

    """ 

    def request_certificate(self, csr, request_type='pkcs10'): 

        """ 

        :param csr: The certificate signing request. 

        :param request_type: The request type (defaults to ``'pkcs10'``). 

        Submit certificate signing request. 

        The command returns a dict with these possible key/value pairs. 

        Some key/value pairs may be absent. 

        +---------------+---------------+---------------+ 

        |result name    |result type    |comments       | 

        +===============+===============+===============+ 

        |serial_number  |unicode [1]_   |               | 

        +---------------+---------------+---------------+ 

        |certificate    |unicode [2]_   |               | 

        +---------------+---------------+---------------+ 

        |request_id     |unicode        |               | 

        +---------------+---------------+---------------+ 

        |subject        |unicode        |               | 

        +---------------+---------------+---------------+ 

        .. [1] Passed through XMLRPC as decimal string. Can convert to 

               optimal integer type (int or long) via int(serial_number) 

        .. [2] Base64 encoded 

        """ 

        try: 

            config = api.Command['config_show']()['result'] 

            subject_base = config.get('ipacertificatesubjectbase')[0] 

            hostname = get_csr_hostname(csr) 

            base = re.split(',\s*(?=\w+=)', subject_base) 

            base.insert(0,'CN=%s' % hostname) 

            subject_base = ",".join(base) 

            request = pkcs10.load_certificate_request(csr) 

            # python-nss normalizes the request subject 

            request_subject = str(pkcs10.get_subject(request)) 

            if str(subject_base).lower() != request_subject.lower(): 

                raise errors.CertificateOperationError(error=_('Request subject "%(request_subject)s" does not match the form "%(subject_base)s"') % \ 

                {'request_subject' : request_subject, 'subject_base' : subject_base}) 

        except errors.CertificateOperationError, e: 

            raise e 

        except NSPRError, e: 

            raise errors.CertificateOperationError(error=_('unable to decode csr: %s') % e) 

        # certutil wants the CSR to have have a header and footer. Add one 

        # if it isn't there. 

        s = csr.find('-----BEGIN NEW CERTIFICATE REQUEST-----') 

        if s == -1: 

            s = csr.find('-----BEGIN CERTIFICATE REQUEST-----') 

            if s == -1: 

                csr = '-----BEGIN NEW CERTIFICATE REQUEST-----\n' + csr + \ 

                      '\n-----END NEW CERTIFICATE REQUEST-----\n' 

        try: 

            (csr_fd, csr_name) = tempfile.mkstemp() 

            os.write(csr_fd, csr) 

            os.close(csr_fd) 

        except Exception, e: 

            try: 

                os.remove(csr_name) 

            except: 

                pass 

            self.log.error('unable to create temporary csr file: %s' % e) 

            raise errors.CertificateOperationError(error=_('file operation')) 

        try: 

            (cert_fd, cert_name) = tempfile.mkstemp() 

            os.close(cert_fd) 

        except Exception, e: 

            try: 

                os.remove(csr_name) 

            except: 

                pass 

            try: 

                os.remove(cert_name) 

            except: 

                pass 

            self.log.error('unable to create temporary certificate file: %s' % e) 

            raise errors.CertificateOperationError(error=_('file operation')) 

        try: 

            serialno = certs.next_serial(self.serial_file) 

        except Exception, e: 

            try: 

                os.remove(csr_name) 

            except: 

                pass 

            try: 

                os.remove(cert_name) 

            except: 

                pass 

            self.log.error('next_serial() failed: %s' % e) 

            raise errors.CertificateOperationError(error=_('cannot obtain next serial number')) 

        try: 

            args = [ 

                "/usr/bin/certutil", 

                "-C", 

                "-d", self.sec_dir, 

                "-c", get_ca_nickname(api.env.realm), 

                "-i", csr_name, 

                "-o", cert_name, 

                "-m", str(serialno), 

                "-v", "60", 

                "-1", 

                "-5", 

                "-6", 

                "-a", 

                "-f", self.pwd_file] 

            self.log.debug("issue cert: %s" % str(args)) 

            p = subprocess.Popen(args, 

                stdin=subprocess.PIPE, 

                stdout=subprocess.PIPE, 

                stderr=subprocess.PIPE, close_fds=True) 

            p.stdin.write("0\n1\n2\n3\n9\ny\n") 

            p.stdin.write("0\n9\nn\n") 

            p.stdin.write("1\n9\nn\n") 

            (stdout, stderr) = p.communicate() 

            status = p.returncode 

            self.log.debug("stdout = %s" % stdout) 

            self.log.debug("stderr = %s" % stderr) 

            if status != 0: 

                try: 

                    os.remove(cert_name) 

                except: 

                    pass 

                self.log.error('certutil failed: %s' % stderr) 

                raise errors.CertificateOperationError(error=_('certutil failure')) 

        finally: 

            try: 

                os.remove(csr_name) 

            except: 

                pass 

        try: 

            cert_fd = open(cert_name) 

            cert = cert_fd.read() 

            cert_fd.close() 

        finally: 

            try: 

                os.remove(cert_name) 

            except: 

                pass 

        try: 

            subject = x509.get_subject(cert) 

            serial = x509.get_serial_number(cert) 

        except NSPRError, e: 

            self.log.error('Unable to decode certificate in entry: %s' % str(e)) 

            raise errors.CertificateOperationError(error='Unable to decode certificate in entry: %s' % str(e)) 

        # To make it look like dogtag return just the base64 data. 

        cert = cert.replace('\n','') 

        cert = cert.replace('\r','') 

        s = cert.find('-----BEGIN CERTIFICATE-----') 

        e = cert.find('-----END CERTIFICATE-----') 

        s = s + 27 

        cert = cert[s:e] 

        cmd_result = {} 

        cmd_result['serial_number'] = unicode(serial) # convert long to decimal unicode string 

        cmd_result['serial_number_hex'] = u'0x%X' % serial 

        cmd_result['certificate']   = unicode(cert) 

        cmd_result['subject']       = unicode(subject) 

        return cmd_result 

api.register(ra)