<commit_before>'''
Created on Apr 30, 2010

@author: jnaous
'''

from django.db import models
import os
import binascii
from django.conf import settings
from datetime import timedelta, datetime
import time
from expedient.common.utils.transport import TestClientTransport
from urlparse import urlparse
from django.contrib.auth.models import User
import xmlrpclib

def get_max_password_len():
    # M2Crypto does not like it when header fields are large. Creates bad
    # requests. So limit to 40 hex characters = 20 bytes = 160 bits.
    return 40

def random_password():
    return binascii.b2a_hex(os.urandom(get_max_password_len()/2))

def add_basic_auth(uri, username=None, password=<PASSWORD>):
    parsed = urlparse(uri.lower())
    if username:
        if password:
            new_url = "%s://%s:%s@%s%s" % (parsed.scheme,
                                           username,
                                           password,
                                           parsed.netloc,
                                           parsed.path)
        else:
            new_url = "%s://%s@%s%s" % (parsed.scheme,
                                        username,
                                        parsed.netloc,
                                        parsed.path)
    else:
        new_url = uri
    return new_url

class BasicAuthServerProxy(xmlrpclib.ServerProxy):
    def __init__(self, uri, username=None, password=None, **kwargs):
        new_url = add_basic_auth(uri, username, password)
        
        # M2Crypto fails when using unicode URLs.
        xmlrpclib.ServerProxy.__init__(self, str(new_url), **kwargs)

    def __repr__(self):
        parsed = urlparse(self._ServerProxy__host)
        new_url = "%s://%s%s" % (parsed.scheme,
                                 parsed.netloc,
                                 parsed.path)
        return (
            "<ServerProxy for %s%s>" %
            (new_url, self._ServerProxy__handler)
            )

    __str__ = __repr__

class PasswordXMLRPCServerProxy(models.Model):
    '''
    Implements a server proxy for XML-RPC calls that checks SSL certs and
    uses a password to talk to the server. The password is automatically renewed
    whenever it expires.
    '''
    
    username = models.CharField(max_length=100)
    password = models.CharField(max_length=get_max_password_len(),
                                default=<PASSWORD>)
    max_password_age = models.IntegerField(
        'Max Password age (days)', default=60)
    password_timestamp = models.DateTimeField(auto_now_add=True)
    url = models.CharField("Server URL", max_length=1024)
    
    verify_certs = models.BooleanField("Verify Certificates?", default=False)
    
    def _reset_proxy(self):
        parsed = urlparse(self.url.lower())
        self.transport = None
        # This scheme is used for debugging and looping back
        if parsed.scheme == "test":
            self.proxy = BasicAuthServerProxy(parsed.path,
                                              username=self.username,
                                              password=<PASSWORD>,
                                              transport=TestClientTransport())
        elif parsed.scheme == "https":
            from M2Crypto.m2xmlrpclib import SSL_Transport
            from M2Crypto.SSL import Context
            self.transport = SSL_Transport(Context(protocol='tlsv1'))
            self.proxy = BasicAuthServerProxy(self.url,
                                              username=self.username,
                                              password=<PASSWORD>,
                                              transport=self.transport)
            self.set_verify_certs()
        else:
            self.proxy = BasicAuthServerProxy(self.url)
    
    def _check_expiry(self):
        # if the password has expired, it's time to set a new one
        if self.password_timestamp:
            max_age = timedelta(days=self.max_password_age)
            expiry_time = self.password_timestamp + max_age
            # normalize because django is screwy
            expiry_time = time.mktime(expiry_time.timetuple())
            now = time.time()
            if expiry_time <= now:
                self.change_password(random_password())
                
    def __init__(self, *args, **kwargs):
        super(PasswordXMLRPCServerProxy, self).__init__(*args, **kwargs)
        self._reset_proxy()
        self._check_expiry()
    
    def __getattr__(self, name):
        if name == "proxy":
            raise AttributeError("Attribute 'proxy' not found.")
        return getattr(self.proxy, name)
        
    def save(self, *args, **kwargs):
        super(PasswordXMLRPCServerProxy, self).save(*args, **kwargs)
        self.set_verify_certs()
        
    def set_verify_certs(self):
        '''Enable/disable SSL certificate verification.'''
        if self.transport:
            from M2Crypto import SSL
            if self.verify_certs:
                self.transport.ssl_ctx.set_verify(
                    SSL.verify_peer | SSL.verify_fail_if_no_peer_cert, 16)
                self.transport.ssl_ctx.load_verify_locations(
                    capath=settings.XMLRPC_TRUSTED_CA_PATH)<commit_msg>don't init the proxy if there's no url<commit_after>'''
Created on Apr 30, 2010

@author: jnaous
'''

from django.db import models
import os
import binascii
from django.conf import settings
from datetime import timedelta, datetime
import time
from expedient.common.utils.transport import TestClientTransport
from urlparse import urlparse
from django.contrib.auth.models import User
import xmlrpclib

def get_max_password_len():
    # M2Crypto does not like it when header fields are large. Creates bad
    # requests. So limit to 40 hex characters = 20 bytes = 160 bits.
    return 40

def random_password():
    return binascii.b2a_hex(os.urandom(get_max_password_len()/2))

def add_basic_auth(uri, username=None, password=None):
    parsed = urlparse(uri.lower())
    if username:
        if password:
            new_url = "%s://%s:%s@%s%s" % (parsed.scheme,
                                           username,
                                           password,
                                           parsed.netloc,
                                           parsed.path)
        else:
            new_url = "%s://%s@%s%s" % (parsed.scheme,
                                        username,
                                        parsed.netloc,
                                        parsed.path)
    else:
        new_url = uri
    return new_url

class BasicAuthServerProxy(xmlrpclib.ServerProxy):
    def __init__(self, uri, username=None, password=<PASSWORD>, **kwargs):
        new_url = add_basic_auth(uri, username, password)
        
        # M2Crypto fails when using unicode URLs.
        xmlrpclib.ServerProxy.__init__(self, str(new_url), **kwargs)

    def __repr__(self):
        parsed = urlparse(self._ServerProxy__host)
        new_url = "%s://%s%s" % (parsed.scheme,
                                 parsed.netloc,
                                 parsed.path)
        return (
            "<ServerProxy for %s%s>" %
            (new_url, self._ServerProxy__handler)
            )

    __str__ = __repr__

class PasswordXMLRPCServerProxy(models.Model):
    '''
    Implements a server proxy for XML-RPC calls that checks SSL certs and
    uses a password to talk to the server. The password is automatically renewed
    whenever it expires.
    '''
    
    username = models.CharField(max_length=100)
    password = models.CharField(max_length=get_max_password_len(),
                                default=<PASSWORD>)
    max_password_age = models.IntegerField(
        'Max Password age (days)', default=60)
    password_timestamp = models.DateTimeField(auto_now_add=True)
    url = models.CharField("Server URL", max_length=1024)
    
    verify_certs = models.BooleanField("Verify Certificates?", default=False)
    
    def _reset_proxy(self):
        parsed = urlparse(self.url.lower())
        self.transport = None
        # This scheme is used for debugging and looping back
        if parsed.scheme == "test":
            self.proxy = BasicAuthServerProxy(parsed.path,
                                              username=self.username,
                                              password=<PASSWORD>,
                                              transport=TestClientTransport())
        elif parsed.scheme == "https":
            from M2Crypto.m2xmlrpclib import SSL_Transport
            from M2Crypto.SSL import Context
            self.transport = SSL_Transport(Context(protocol='tlsv1'))
            self.proxy = BasicAuthServerProxy(self.url,
                                              username=self.username,
                                              password=self.password,
                                              transport=self.transport)
            self.set_verify_certs()
        else:
            self.proxy = BasicAuthServerProxy(self.url)
    
    def _check_expiry(self):
        # if the password has expired, it's time to set a new one
        if self.password_timestamp and self.max_password_age:
            max_age = timedelta(days=self.max_password_age)
            expiry_time = self.password_timestamp + max_age
            # normalize because django is screwy
            expiry_time = time.mktime(expiry_time.timetuple())
            now = time.time()
            if expiry_time <= now:
                self.change_password(random_password())
                
    def __init__(self, *args, **kwargs):
        super(PasswordXMLRPCServerProxy, self).__init__(*args, **kwargs)
        if self.url:
            self._reset_proxy()
        self._check_expiry()
    
    def __getattr__(self, name):
        if name == "proxy":
            raise AttributeError("Attribute 'proxy' not found.")
        return getattr(self.proxy, name)
        
    def save(self, *args, **kwargs):
        super(PasswordXMLRPCServerProxy, self).save(*args, **kwargs)
        self.set_verify_certs()
        
    def set_verify_certs(self):
        '''Enable/disable SSL certificate verification.'''
        if self.transport:
            from M2Crypto import SSL
            if self.verify_certs:
                self.transport.ssl_ctx.set_verify(
                    SSL.verify_peer | SSL.verify_fail_if_no_peer_cert, 16)
                self.transport.ssl_ctx.load_verify_locations(
                    capath=settings.XMLRPC_TRUSTED_CA_PATH)
{'fraction_non_alphanumeric': 0.0848761036741669, 'fraction_numerical': 0.006645779929744612, 'mean_word_length': 1.7963897000265463, 'pattern_counts': {'":': 6, '<': 14, '<?xml version=': 0, '>': 12, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 8}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 50, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3746867', 'n_tokens_mistral': 2788, 'n_tokens_neox': 2702, 'n_words': 652}