<s> """<STR_LIT>""" <EOL> from django . contrib . auth . models import AbstractBaseUser <EOL> from django . db import connection <EOL> from django . db import models <EOL> from django . db import transaction <EOL> from django . forms import ValidationError <EOL> from django . utils import timezone <EOL> from functools import wraps <EOL> from wwwhisper_auth import url_utils <EOL> from wwwhisper_auth import email_re <EOL> import logging <EOL> import random <EOL> import re <EOL> import threading <EOL> import uuid as uuidgen <EOL> logger = logging . getLogger ( __name__ ) <EOL> class LimitExceeded ( Exception ) : <EOL> pass <EOL> class ValidatedModel ( models . Model ) : <EOL> """<STR_LIT>""" <EOL> class Meta : <EOL> """<STR_LIT>""" <EOL> abstract = True <EOL> app_label = '<STR_LIT>' <EOL> def save ( self , * args , ** kwargs ) : <EOL> self . full_clean ( ) <EOL> return super ( ValidatedModel , self ) . save ( * args , ** kwargs ) <EOL> SINGLE_SITE_ID = '<STR_LIT>' <EOL> class Site ( ValidatedModel ) : <EOL> """<STR_LIT>""" <EOL> site_id = models . TextField ( primary_key = True , db_index = True , editable = False ) <EOL> mod_id = models . IntegerField ( default = <NUM_LIT:0> ) <EOL> _default_skin = { <EOL> '<STR_LIT:title>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:message>' : '<STR_LIT>' + '<STR_LIT>' , <EOL> } <EOL> title = models . CharField ( max_length = <NUM_LIT> , blank = True ) <EOL> header = models . CharField ( max_length = <NUM_LIT:100> , blank = True ) <EOL> message = models . CharField ( max_length = <NUM_LIT> , blank = True ) <EOL> branding = models . BooleanField ( default = True ) <EOL> aliases_limit = None <EOL> users_limit = None <EOL> locations_limit = None <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( Site , self ) . __init__ ( * args , ** kwargs ) <EOL> self . mod_id_lock = threading . Lock ( ) <EOL> def heavy_init ( self ) : <EOL> """<STR_LIT>""" <EOL> self . locations = LocationsCollection ( self ) <EOL> self . users = UsersCollection ( self ) <EOL> self . aliases = AliasesCollection ( self ) <EOL> def site_modified ( self ) : <EOL> """<STR_LIT>""" <EOL> cursor = connection . cursor ( ) <EOL> cursor . execute ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , [ self . site_id ] ) <EOL> cursor . close ( ) <EOL> mod_id = self . mod_id_from_db ( ) <EOL> with self . mod_id_lock : <EOL> self . mod_id = mod_id <EOL> def skin ( self ) : <EOL> """<STR_LIT>""" <EOL> result = dict ( [ ( attr , getattr ( self , attr ) or self . _default_skin [ attr ] ) <EOL> for attr in self . _default_skin . iterkeys ( ) ] ) <EOL> result [ '<STR_LIT>' ] = self . branding <EOL> return result <EOL> def update_skin ( self , title , header , message , branding ) : <EOL> for attr in self . _default_skin . iterkeys ( ) : <EOL> arg = locals ( ) [ attr ] . strip ( ) <EOL> if arg == self . _default_skin [ attr ] : <EOL> arg = '<STR_LIT>' <EOL> setattr ( self , attr , arg ) <EOL> self . branding = branding <EOL> self . save ( ) <EOL> self . site_modified ( ) <EOL> def get_mod_id_ts ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . mod_id_lock : <EOL> return self . mod_id <EOL> def mod_id_from_db ( self ) : <EOL> """<STR_LIT>""" <EOL> cursor = connection . cursor ( ) <EOL> cursor . execute ( <EOL> '<STR_LIT>' , <EOL> [ self . site_id ] ) <EOL> row = cursor . fetchone ( ) <EOL> cursor . close ( ) <EOL> if row is None : <EOL> return None <EOL> return row [ <NUM_LIT:0> ] <EOL> def modify_site ( decorated_method ) : <EOL> """<STR_LIT>""" <EOL> @ wraps ( decorated_method ) <EOL> def wrapper ( self , * args , ** kwargs ) : <EOL> result = decorated_method ( self , * args , ** kwargs ) <EOL> self . site . site_modified ( ) <EOL> return result <EOL> return wrapper <EOL> class SitesCollection ( object ) : <EOL> def create_item ( self , site_id , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> site = Site . objects . create ( site_id = site_id , ** kwargs ) <EOL> site . heavy_init ( ) <EOL> return site <EOL> def find_item ( self , site_id ) : <EOL> site = _find ( Site , site_id = site_id ) <EOL> if site is not None : <EOL> site . heavy_init ( ) <EOL> return site <EOL> def delete_item ( self , site_id ) : <EOL> site = self . find_item ( site_id ) <EOL> if site is None : <EOL> return False <EOL> site . delete ( ) <EOL> return True <EOL> class User ( AbstractBaseUser ) : <EOL> class Meta : <EOL> app_label = '<STR_LIT>' <EOL> site = models . ForeignKey ( Site , related_name = '<STR_LIT:+>' ) <EOL> uuid = models . CharField ( max_length = <NUM_LIT> , db_index = True , <EOL> editable = False , unique = True ) <EOL> email = models . EmailField ( db_index = True ) <EOL> USERNAME_FIELD = '<STR_LIT>' <EOL> REQUIRED_FIELDS = [ '<STR_LIT:email>' , '<STR_LIT>' ] <EOL> def attributes_dict ( self , site_url ) : <EOL> """<STR_LIT>""" <EOL> return _add_common_attributes ( self , site_url , { '<STR_LIT:email>' : self . email } ) <EOL> @ models . permalink <EOL> def get_absolute_url ( self ) : <EOL> return ( '<STR_LIT>' , ( ) , { '<STR_LIT>' : self . uuid } ) <EOL> class Location ( ValidatedModel ) : <EOL> """<STR_LIT>""" <EOL> OPEN_ACCESS_CHOICES = ( <EOL> ( '<STR_LIT:n>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:y>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:a>' , '<STR_LIT>' ) , <EOL> ) <EOL> site = models . ForeignKey ( Site , related_name = '<STR_LIT:+>' ) <EOL> path = models . TextField ( db_index = True ) <EOL> uuid = models . CharField ( max_length = <NUM_LIT> , db_index = True , <EOL> editable = False , unique = True ) <EOL> open_access = models . CharField ( max_length = <NUM_LIT:2> , choices = OPEN_ACCESS_CHOICES , <EOL> default = '<STR_LIT:n>' ) <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( Location , self ) . __init__ ( * args , ** kwargs ) <EOL> def permissions ( self ) : <EOL> return self . site . locations . get_permissions ( self . id ) <EOL> def __unicode__ ( self ) : <EOL> return "<STR_LIT:%s>" % ( self . path ) <EOL> @ models . permalink <EOL> def get_absolute_url ( self ) : <EOL> """<STR_LIT>""" <EOL> return ( '<STR_LIT>' , ( ) , { '<STR_LIT>' : self . uuid } ) <EOL> @ modify_site <EOL> def grant_open_access ( self , require_login ) : <EOL> """<STR_LIT>""" <EOL> if require_login : <EOL> self . open_access = '<STR_LIT:a>' <EOL> else : <EOL> self . open_access = '<STR_LIT:y>' <EOL> self . save ( ) <EOL> def open_access_granted ( self ) : <EOL> return self . open_access in ( '<STR_LIT:y>' , '<STR_LIT:a>' ) <EOL> def open_access_requires_login ( self ) : <EOL> return self . open_access == '<STR_LIT:a>' <EOL> @ modify_site <EOL> def revoke_open_access ( self ) : <EOL> """<STR_LIT>""" <EOL> self . open_access = '<STR_LIT:n>' <EOL> self . save ( ) <EOL> def can_access ( self , user ) : <EOL> """<STR_LIT>""" <EOL> if user . site_id != self . site_id : <EOL> return False <EOL> return ( self . open_access_granted ( ) <EOL> or self . permissions ( ) . get ( user . id ) != None ) <EOL> @ modify_site <EOL> def grant_access ( self , user_uuid ) : <EOL> """<STR_LIT>""" <EOL> user = self . site . users . find_item ( uuid = user_uuid ) <EOL> if user is None : <EOL> raise LookupError ( '<STR_LIT>' ) <EOL> permission = self . permissions ( ) . get ( user . id ) <EOL> created = False <EOL> if permission is None : <EOL> created = True <EOL> permission = Permission . objects . create ( <EOL> http_location_id = self . id , user_id = user . id , site_id = self . site_id ) <EOL> return ( permission , created ) <EOL> @ modify_site <EOL> def revoke_access ( self , user_uuid ) : <EOL> """<STR_LIT>""" <EOL> permission = self . get_permission ( user_uuid ) <EOL> permission . delete ( ) <EOL> def get_permission ( self , user_uuid ) : <EOL> """<STR_LIT>""" <EOL> user = self . site . users . find_item ( uuid = user_uuid ) <EOL> if user is None : <EOL> raise LookupError ( '<STR_LIT>' ) <EOL> permission = self . permissions ( ) . get ( user . id ) <EOL> if permission is None : <EOL> raise LookupError ( '<STR_LIT>' ) <EOL> return permission <EOL> def allowed_users ( self ) : <EOL> """<STR_LIT>""" <EOL> return [ self . site . users . find_item_by_pk ( user_id ) <EOL> for user_id in self . permissions ( ) . iterkeys ( ) ] <EOL> def attributes_dict ( self , site_url ) : <EOL> """<STR_LIT>""" <EOL> result = { <EOL> '<STR_LIT:path>' : self . path , <EOL> '<STR_LIT>' : [ <EOL> user . attributes_dict ( site_url ) for user in self . allowed_users ( ) <EOL> ] , <EOL> } <EOL> if self . open_access_granted ( ) : <EOL> result [ '<STR_LIT>' ] = { <EOL> '<STR_LIT>' : self . open_access_requires_login ( ) <EOL> } <EOL> return _add_common_attributes ( self , site_url , result ) <EOL> class Permission ( ValidatedModel ) : <EOL> """<STR_LIT>""" <EOL> http_location = models . ForeignKey ( Location , related_name = '<STR_LIT:+>' ) <EOL> site = models . ForeignKey ( Site , related_name = '<STR_LIT:+>' ) <EOL> user = models . ForeignKey ( User , related_name = '<STR_LIT:+>' ) <EOL> def __unicode__ ( self ) : <EOL> return "<STR_LIT>" % ( self . http_location , self . user . email ) <EOL> @ models . permalink <EOL> def get_absolute_url ( self ) : <EOL> """<STR_LIT>""" <EOL> return ( '<STR_LIT>' , ( ) , <EOL> { '<STR_LIT>' : self . http_location . uuid , <EOL> '<STR_LIT>' : self . user . uuid } ) <EOL> def attributes_dict ( self , site_url ) : <EOL> """<STR_LIT>""" <EOL> return _add_common_attributes ( <EOL> self , site_url , { '<STR_LIT:user>' : self . user . attributes_dict ( site_url ) } ) <EOL> class Alias ( ValidatedModel ) : <EOL> """<STR_LIT>""" <EOL> site = models . ForeignKey ( Site , related_name = '<STR_LIT:+>' ) <EOL> url = models . TextField ( db_index = True ) <EOL> uuid = models . CharField ( max_length = <NUM_LIT> , db_index = True , <EOL> editable = False , unique = True ) <EOL> force_ssl = models . BooleanField ( default = False ) <EOL> @ models . permalink <EOL> def get_absolute_url ( self ) : <EOL> return ( '<STR_LIT>' , ( ) , { '<STR_LIT>' : self . uuid } ) <EOL> def attributes_dict ( self , site_url ) : <EOL> return _add_common_attributes ( self , site_url , { '<STR_LIT:url>' : self . url } ) <EOL> class Collection ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , site ) : <EOL> self . site = site <EOL> self . update_cache ( ) <EOL> def update_cache ( self ) : <EOL> self . _cached_items_dict = { } <EOL> self . _cached_items_list = [ ] <EOL> for item in self . model_class . objects . filter ( site_id = self . site . site_id ) : <EOL> self . _cached_items_dict [ item . id ] = item <EOL> self . _cached_items_list . append ( item ) <EOL> item . site = self . site <EOL> self . cache_mod_id = self . site . mod_id <EOL> def is_cache_obsolete ( self ) : <EOL> return self . site . mod_id != self . cache_mod_id <EOL> def all ( self ) : <EOL> if self . is_cache_obsolete ( ) : <EOL> self . update_cache ( ) <EOL> return self . _cached_items_list <EOL> def all_dict ( self ) : <EOL> if self . is_cache_obsolete ( ) : <EOL> self . update_cache ( ) <EOL> return self . _cached_items_dict <EOL> def count ( self ) : <EOL> return len ( self . all ( ) ) <EOL> def get_unique ( self , filter_fun ) : <EOL> """<STR_LIT>""" <EOL> result = filter ( filter_fun , self . all ( ) ) <EOL> count = len ( result ) <EOL> assert count <= <NUM_LIT:1> <EOL> if count == <NUM_LIT:0> : <EOL> return None <EOL> return result [ <NUM_LIT:0> ] <EOL> def find_item ( self , uuid ) : <EOL> return self . get_unique ( lambda item : item . uuid == uuid ) <EOL> def find_item_by_pk ( self , pk ) : <EOL> return self . all_dict ( ) . get ( pk , None ) <EOL> @ modify_site <EOL> def delete_item ( self , uuid ) : <EOL> """<STR_LIT>""" <EOL> item = self . find_item ( uuid ) <EOL> if item is None : <EOL> return False <EOL> item . delete ( ) <EOL> return True <EOL> def _do_create_item ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> item = self . model_class . objects . create ( <EOL> site = self . site , uuid = str ( uuidgen . uuid4 ( ) ) , ** kwargs ) <EOL> item . site = self . site <EOL> return item <EOL> class UsersCollection ( Collection ) : <EOL> """<STR_LIT>""" <EOL> item_name = '<STR_LIT:user>' <EOL> model_class = User <EOL> @ modify_site <EOL> def create_item ( self , email ) : <EOL> """<STR_LIT>""" <EOL> users_limit = self . site . users_limit <EOL> if ( users_limit is not None and self . count ( ) >= users_limit ) : <EOL> raise LimitExceeded ( '<STR_LIT>' ) <EOL> encoded_email = _encode_email ( email ) <EOL> if encoded_email is None : <EOL> raise ValidationError ( '<STR_LIT>' ) <EOL> if self . find_item_by_email ( encoded_email ) is not None : <EOL> raise ValidationError ( '<STR_LIT>' ) <EOL> return self . _do_create_item ( email = encoded_email , <EOL> last_login = timezone . now ( ) ) <EOL> def find_item_by_email ( self , email ) : <EOL> encoded_email = _encode_email ( email ) <EOL> if encoded_email is None : <EOL> return None <EOL> return self . get_unique ( lambda user : user . email == encoded_email ) <EOL> class LocationsCollection ( Collection ) : <EOL> """<STR_LIT>""" <EOL> PATH_LEN_LIMIT = <NUM_LIT> <EOL> item_name = '<STR_LIT:location>' <EOL> model_class = Location <EOL> def update_cache ( self ) : <EOL> super ( LocationsCollection , self ) . update_cache ( ) <EOL> self . _cached_permissions = { } <EOL> for p in Permission . objects . filter ( site = self . site ) : <EOL> self . _cached_permissions . setdefault ( <EOL> p . http_location_id , { } ) [ p . user_id ] = p <EOL> def get_permissions ( self , location_id ) : <EOL> """<STR_LIT>""" <EOL> if self . is_cache_obsolete ( ) : <EOL> self . update_cache ( ) <EOL> return self . _cached_permissions . get ( location_id , { } ) <EOL> @ modify_site <EOL> def create_item ( self , path ) : <EOL> """<STR_LIT>""" <EOL> locations_limit = self . site . locations_limit <EOL> if ( locations_limit is not None and self . count ( ) >= locations_limit ) : <EOL> raise LimitExceeded ( '<STR_LIT>' ) <EOL> if not url_utils . is_canonical ( path ) : <EOL> raise ValidationError ( <EOL> '<STR_LIT>' '<STR_LIT>' ) <EOL> if len ( path ) > self . PATH_LEN_LIMIT : <EOL> raise ValidationError ( '<STR_LIT>' ) <EOL> if url_utils . contains_fragment ( path ) : <EOL> raise ValidationError ( <EOL> "<STR_LIT>" ) <EOL> if url_utils . contains_query ( path ) : <EOL> raise ValidationError ( <EOL> "<STR_LIT>" ) <EOL> if url_utils . contains_params ( path ) : <EOL> raise ValidationError ( <EOL> "<STR_LIT>" ) <EOL> try : <EOL> path . encode ( '<STR_LIT:ascii>' ) <EOL> except UnicodeError : <EOL> raise ValidationError ( <EOL> '<STR_LIT>' ) <EOL> if self . get_unique ( lambda item : item . path == path ) is not None : <EOL> raise ValidationError ( '<STR_LIT>' ) <EOL> return self . _do_create_item ( path = path ) <EOL> def find_location ( self , canonical_path ) : <EOL> """<STR_LIT>""" <EOL> canonical_path_len = len ( canonical_path ) <EOL> longest_matched_location = None <EOL> longest_matched_location_len = - <NUM_LIT:1> <EOL> for location in self . all ( ) : <EOL> probed_path = location . path <EOL> probed_path_len = len ( probed_path ) <EOL> trailing_slash_index = None <EOL> if probed_path [ probed_path_len - <NUM_LIT:1> ] == '<STR_LIT:/>' : <EOL> trailing_slash_index = probed_path_len - <NUM_LIT:1> <EOL> else : <EOL> trailing_slash_index = probed_path_len <EOL> if ( canonical_path . startswith ( probed_path ) and <EOL> probed_path_len > longest_matched_location_len and <EOL> ( probed_path_len == canonical_path_len or <EOL> canonical_path [ trailing_slash_index ] == '<STR_LIT:/>' ) ) : <EOL> longest_matched_location_len = probed_path_len <EOL> longest_matched_location = location <EOL> return longest_matched_location <EOL> def has_open_location_with_login ( self ) : <EOL> for location in self . all ( ) : <EOL> if ( location . open_access_granted ( ) and <EOL> location . open_access_requires_login ( ) ) : <EOL> return True <EOL> return False <EOL> def has_open_location ( self ) : <EOL> for location in self . all ( ) : <EOL> if location . open_access_granted ( ) : <EOL> return True <EOL> return False <EOL> class AliasesCollection ( Collection ) : <EOL> item_name = '<STR_LIT>' <EOL> model_class = Alias <EOL> ALIAS_LEN_LIMIT = <NUM_LIT:8> + <NUM_LIT> + <NUM_LIT:6> <EOL> @ modify_site <EOL> def create_item ( self , url ) : <EOL> aliases_limit = self . site . aliases_limit <EOL> if ( aliases_limit is not None and self . count ( ) >= aliases_limit ) : <EOL> raise LimitExceeded ( '<STR_LIT>' ) <EOL> if len ( url ) > self . ALIAS_LEN_LIMIT : <EOL> raise ValidationError ( '<STR_LIT>' ) <EOL> url = url . strip ( ) . lower ( ) <EOL> ( valid , error ) = url_utils . validate_site_url ( url ) <EOL> if not valid : <EOL> raise ValidationError ( '<STR_LIT>' + error ) <EOL> url = url_utils . remove_default_port ( url ) <EOL> if self . find_item_by_url ( url ) : <EOL> raise ValidationError ( '<STR_LIT>' ) <EOL> return self . _do_create_item ( url = url ) <EOL> def find_item_by_url ( self , url ) : <EOL> return self . get_unique ( lambda item : item . url == url ) <EOL> def _uuid2urn ( uuid ) : <EOL> return '<STR_LIT>' + uuid <EOL> def _add_common_attributes ( item , site_url , attributes_dict ) : <EOL> """<STR_LIT>""" <EOL> attributes_dict [ '<STR_LIT>' ] = site_url + item . get_absolute_url ( ) <EOL> if hasattr ( item , '<STR_LIT>' ) : <EOL> attributes_dict [ '<STR_LIT:id>' ] = _uuid2urn ( item . uuid ) <EOL> return attributes_dict <EOL> def _find ( model_class , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> items = [ item for item in model_class . objects . filter ( ** kwargs ) ] <EOL> count = len ( items ) <EOL> assert count <= <NUM_LIT:1> <EOL> if count == <NUM_LIT:0> : <EOL> return None <EOL> return items [ <NUM_LIT:0> ] <EOL> def _encode_email ( email ) : <EOL> """<STR_LIT>""" <EOL> encoded_email = email . lower ( ) <EOL> if not is_email_valid ( encoded_email ) : <EOL> return None <EOL> return encoded_email <EOL> def is_email_valid ( email ) : <EOL> return re . match ( email_re . EMAIL_VALIDATION_RE , email ) </s>
