kallithea Changeset - 6c28533d122c

Changeset - 6c28533d122c

Parent rev.

Child rev.

[Not reviewed]

beta

0 3 0

Marcin Kuzminski - 13 years ago 2013-01-23 23:51:57
marcin@python-works.com

IP restrictions now also enabled for IPv6

3 files changed with 34 insertions and 24 deletions:

rhodecode/lib/auth.py

rhodecode/model/db.py

rhodecode/model/validators.py

0 comments (0 inline, 0 general)

rhodecode/lib/auth.py

➞

Show inline comments

@@ @@ -25,24 +25,25 @@ @@
 import random
 import logging
 import traceback
 import hashlib
 from tempfile import _RandomNameSequence
 from decorator import decorator
 from pylons import config, url, request
 from pylons.controllers.util import abort, redirect
 from pylons.i18n.translation import _
 from sqlalchemy.orm.exc import ObjectDeletedError
 from rhodecode import __platform__, is_windows, is_unix
 from rhodecode.model.meta import Session
 from rhodecode.lib.utils2 import str2bool, safe_unicode
 from rhodecode.lib.exceptions import LdapPasswordError, LdapUsernameError
 from rhodecode.lib.utils import get_repo_slug, get_repos_group_slug
 from rhodecode.lib.auth_ldap import AuthLdap
 from rhodecode.model import meta
 from rhodecode.model.user import UserModel
 from rhodecode.model.db import Permission, RhodeCodeSetting, User, UserIpMap
@@ @@ -438,26 +439,31 @@ class AuthUser(object): @@
         username = cookie_store.get('username')
         api_key = cookie_store.get('api_key')
         return AuthUser(user_id, api_key, username)
     @classmethod
     def get_allowed_ips(cls, user_id, cache=False):
         _set = set()
         user_ips = UserIpMap.query().filter(UserIpMap.user_id == user_id)
         if cache:
             user_ips = user_ips.options(FromCache("sql_cache_short",
                                                   "get_user_ips_%s" % user_id))
         for ip in user_ips:
             try:
             _set.add(ip.ip_addr)
         return _set or set(['0.0.0.0/0'])
             except ObjectDeletedError:
                 # since we use heavy caching sometimes it happens that we get
                 # deleted objects here, we just skip them
                 pass
         return _set or set(['0.0.0.0/0', '::/0'])
 def set_available_permissions(config):
     """
     This function will propagate pylons globals with all available defined
     permission given in db. We don't want to check each time from db for new
     permissions since adding a new permission also requires application restart
     ie. to decorate new views with the newly created permission
     :param config: current pylons config instance
     """
@@ @@ -981,15 +987,22 @@ class HasRepoPermissionAnyApi(_BaseApiPe @@
 def check_ip_access(source_ip, allowed_ips=None):
     """
     Checks if source_ip is a subnet of any of allowed_ips.
     :param source_ip:
     :param allowed_ips: list of allowed ips together with mask
     """
     from rhodecode.lib import ipaddr
     log.debug('checking if ip:%s is subnet of %s' % (source_ip, allowed_ips))
     if isinstance(allowed_ips, (tuple, list, set)):
         for ip in allowed_ips:
             try:
             if ipaddr.IPAddress(source_ip) in ipaddr.IPNetwork(ip):
                 return True
                 # for any case we cannot determine the IP, don't crash just
                 # skip it and log as error, we want to say forbidden still when
                 # sending bad IP
             except Exception:
                 log.error(traceback.format_exc())
                 continue
     return False

rhodecode/model/db.py

➞

Show inline comments

@@ @@ -553,25 +553,25 @@ class UserIpMap(Base, BaseModel): @@
+    )
     __mapper_args__ = {}
     ip_id = Column("ip_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)
     user_id = Column("user_id", Integer(), ForeignKey('users.user_id'), nullable=True, unique=None, default=None)
     ip_addr = Column("ip_addr", String(255, convert_unicode=False, assert_unicode=None), nullable=True, unique=False, default=None)
     active = Column("active", Boolean(), nullable=True, unique=None, default=True)
     user = relationship('User', lazy='joined')
     @classmethod
     def _get_ip_range(cls, ip_addr):
         from rhodecode.lib import ipaddr
-        net = ipaddr.IPv4Network(ip_addr)
+        net = ipaddr.IPNetwork(address=ip_addr)
         return [str(net.network), str(net.broadcast)]
     def __json__(self):
         return dict(
           ip_addr=self.ip_addr,
           ip_range=self._get_ip_range(self.ip_addr)
+        )
 class UserLog(Base, BaseModel):
     __tablename__ = 'user_logs'
     __table_args__ = (

rhodecode/model/validators.py

➞

Show inline comments

@@ @@ -5,24 +5,25 @@ import os @@
 import re
 import formencode
 import logging
 from collections import defaultdict
 from pylons.i18n.translation import _
 from webhelpers.pylonslib.secure_form import authentication_token
 from formencode.validators import (
     UnicodeString, OneOf, Int, Number, Regex, Email, Bool, StringBoolean, Set,
     NotEmpty, IPAddress, CIDR
+)
 from rhodecode.lib.compat import OrderedSet
 from rhodecode.lib import ipaddr
 from rhodecode.lib.utils import repo_name_slug
 from rhodecode.model.db import RepoGroup, Repository, UsersGroup, User,\
     ChangesetStatus
 from rhodecode.lib.exceptions import LdapImportError
 from rhodecode.config.routing import ADMIN_PREFIX
 from rhodecode.lib.auth import HasReposGroupPermissionAny
 # silence warnings and pylint
 UnicodeString, OneOf, Int, Number, Regex, Email, Bool, StringBoolean, Set, \
     NotEmpty, IPAddress, CIDR
 log = logging.getLogger(__name__)
@@ @@ -702,44 +703,40 @@ def NotReviewedRevisions(repo_id): @@
                 revs = ','.join([x[1] for x in errors])
                 msg = M(self, 'rev_already_reviewed', state, revs=revs)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(revisions=revs)
+                )
     return _validator
 def ValidIp():
     class _validator(CIDR):
         messages = dict(
             badFormat=_('Please enter a valid IP address (a.b.c.d)'),
             illegalOctets=_('The octets must be within the range of 0-255'
                 ' (not %(octet)r)'),
             badFormat=_('Please enter a valid IPv4 or IpV6 address'),
             illegalBits=_('The network size (bits) must be within the range'
                 ' of 0-32 (not %(bits)r)'))
         def to_python(self, value, state):
             v = super(_validator, self).to_python(value, state)
             v = v.strip()
             net = ipaddr.IPNetwork(address=v)
             if isinstance(net, ipaddr.IPv4Network):
                 #if IPv4 doesn't end with a mask, add /32
                 if '/' not in value:
                     v += '/32'
             if isinstance(net, ipaddr.IPv6Network):
                 #if IPv6 doesn't end with a mask, add /128
                 if '/' not in value:
                     v += '/128'
             return v
         def validate_python(self, value, state):
             try:
                 # Split into octets and bits
                 if '/' in value:  # a.b.c.d/e
                     addr, bits = value.split('/')
                 else:  # a.b.c.d
                     addr, bits = value, 32
                 # Use IPAddress validator to validate the IP part
                 IPAddress.validate_python(self, addr, state)
                 # Bits (netmask) correct?
                 if not 0 <= int(bits) <= 32:
                     raise formencode.Invalid(
                         self.message('illegalBits', state, bits=bits),
                         value, state)
             # Splitting faild: wrong syntax
                 addr = value.strip()
                 #this raises an ValueError if address is not IpV4 or IpV6
                 ipaddr.IPNetwork(address=addr)
             except ValueError:
                 raise formencode.Invalid(self.message('badFormat', state),
                                          value, state)
         def to_python(self, value, state):
             v = super(_validator, self).to_python(value, state)
             #if IP doesn't end with a mask, add /32
             if '/' not in value:
                 v += '/32'
             return v
     return _validator

0 comments (0 inline, 0 general)