engineering/unrhodecode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

gists: use colander schema to validate input data.

Browse source 
- brings consistent validation acros API and web
- use nicer and stricter schemas to validate data
- fixes #4118
This commit is contained in:
Marcin Kuzminski 2016-07-31 20:11:12 +02:00
parent 95e32917c7
commit 198c6fda61
27 changed files with 773 additions and 237 deletions
Download patch file Download diff file Expand all files Collapse all files

8
rhodecode/api/__init__.py
View file

@ -30,7 +30,8 @@ from pyramid.renderers import render
from pyramid.response import Response
from pyramid.httpexceptions import HTTPNotFound
from rhodecode.api.exc import JSONRPCBaseError, JSONRPCError, JSONRPCForbidden
from rhodecode.api.exc import (
JSONRPCBaseError, JSONRPCError, JSONRPCForbidden, JSONRPCValidationError)
from rhodecode.lib.auth import AuthUser
from rhodecode.lib.base import get_ip_addr
from rhodecode.lib.ext_json import json
@ -127,6 +128,11 @@ def exception_view(exc, request):
if isinstance(exc, JSONRPCError):
fault_message = exc.message
log.debug('json-rpc error rpc_id:%s "%s"', rpc_id, fault_message)
elif isinstance(exc, JSONRPCValidationError):
colander_exc = exc.colander_exception
#TODO: think maybe of nicer way to serialize errors ?
fault_message = colander_exc.asdict()
log.debug('json-rpc error rpc_id:%s "%s"', rpc_id, fault_message)
elif isinstance(exc, JSONRPCForbidden):
fault_message = 'Access was denied to this resource.'
log.warning('json-rpc forbidden call rpc_id:%s "%s"', rpc_id, fault_message)

8
rhodecode/api/exc.py
View file

@ -27,5 +27,13 @@ class JSONRPCError(JSONRPCBaseError):
pass
class JSONRPCValidationError(JSONRPCBaseError):
def __init__(self, *args, **kwargs):
self.colander_exception = kwargs.pop('colander_exc')
super(JSONRPCValidationError, self).__init__(*args, **kwargs)
class JSONRPCForbidden(JSONRPCBaseError):
pass

28
rhodecode/api/tests/test_create_gist.py
View file

@ -43,7 +43,7 @@ class TestApiCreateGist(object):
description='foobar-gist',
gist_type=gist_type,
acl_level=gist_acl_level,
files={'foobar': {'content': 'foo'}})
files={'foobar_ąć': {'content': 'foo'}})
response = api_call(self.app, params)
response_json = response.json
gist = response_json['result']['gist']
@ -68,6 +68,32 @@ class TestApiCreateGist(object):
finally:
Fixture().destroy_gists()
@pytest.mark.parametrize("expected, lifetime, gist_type, gist_acl_level, files", [
({'gist_type': '"ups" is not one of private, public'},
10, 'ups', Gist.ACL_LEVEL_PUBLIC, {'f': {'content': 'f'}}),
({'lifetime': '-120 is less than minimum value -1'},
-120, Gist.GIST_PUBLIC, Gist.ACL_LEVEL_PUBLIC, {'f': {'content': 'f'}}),
({'0.content': 'Required'},
10, Gist.GIST_PUBLIC, Gist.ACL_LEVEL_PUBLIC, {'f': {'x': 'f'}}),
])
def test_api_try_create_gist(
self, expected, lifetime, gist_type, gist_acl_level, files):
id_, params = build_data(
self.apikey_regular, 'create_gist',
lifetime=lifetime,
description='foobar-gist',
gist_type=gist_type,
acl_level=gist_acl_level,
files=files)
response = api_call(self.app, params)
try:
assert_error(id_, expected, given=response.body)
finally:
Fixture().destroy_gists()
@mock.patch.object(GistModel, 'create', crash)
def test_api_create_gist_exception_occurred(self):
id_, params = build_data(self.apikey_regular, 'create_gist', files={})

2
rhodecode/api/utils.py
View file

@ -34,8 +34,6 @@ from rhodecode.lib.vcs.exceptions import RepositoryError
log = logging.getLogger(__name__)
class OAttr(object):
"""
Special Option that defines other attribute, and can default to them

61
rhodecode/api/views/gist_api.py
View file

@ -23,6 +23,7 @@ import logging
import time
from rhodecode.api import jsonrpc_method, JSONRPCError
from rhodecode.api.exc import JSONRPCValidationError
from rhodecode.api.utils import (
Optional, OAttr, get_gist_or_error, get_user_or_error,
has_superadmin_permission)
@ -96,7 +97,8 @@ def get_gists(request, apiuser, userid=Optional(OAttr('apiuser'))):
@jsonrpc_method()
def create_gist(
request, apiuser, files, owner=Optional(OAttr('apiuser')),
request, apiuser, files, gistid=Optional(None),
owner=Optional(OAttr('apiuser')),
gist_type=Optional(Gist.GIST_PUBLIC), lifetime=Optional(-1),
acl_level=Optional(Gist.ACL_LEVEL_PUBLIC),
description=Optional('')):
@ -108,10 +110,11 @@ def create_gist(
:param files: files to be added to the gist. The data structure has
to match the following example::
{'filename': {'content':'...', 'lexer': null},
'filename2': {'content':'...', 'lexer': null}}
{'filename1': {'content':'...'}, 'filename2': {'content':'...'}}
:type files: dict
:param gistid: Set a custom id for the gist
:type gistid: Optional(str)
:param owner: Set the gist owner, defaults to api method caller
:type owner: Optional(str or int)
:param gist_type: type of gist ``public`` or ``private``
@ -148,23 +151,49 @@ def create_gist(
}
"""
from rhodecode.model import validation_schema
from rhodecode.model.validation_schema.schemas import gist_schema
if isinstance(owner, Optional):
owner = apiuser.user_id
owner = get_user_or_error(owner)
lifetime = Optional.extract(lifetime)
schema = gist_schema.GistSchema().bind(
# bind the given values if it's allowed, however the deferred
# validator will still validate it according to other rules
lifetime_options=[lifetime])
try:
if isinstance(owner, Optional):
owner = apiuser.user_id
nodes = gist_schema.nodes_to_sequence(
files, colander_node=schema.get('nodes'))
owner = get_user_or_error(owner)
description = Optional.extract(description)
gist_type = Optional.extract(gist_type)
lifetime = Optional.extract(lifetime)
acl_level = Optional.extract(acl_level)
schema_data = schema.deserialize(dict(
gistid=Optional.extract(gistid),
description=Optional.extract(description),
gist_type=Optional.extract(gist_type),
lifetime=lifetime,
gist_acl_level=Optional.extract(acl_level),
nodes=nodes
))
gist = GistModel().create(description=description,
owner=owner,
gist_mapping=files,
gist_type=gist_type,
lifetime=lifetime,
gist_acl_level=acl_level)
# convert to safer format with just KEYs so we sure no duplicates
schema_data['nodes'] = gist_schema.sequence_to_nodes(
schema_data['nodes'], colander_node=schema.get('nodes'))
except validation_schema.Invalid as err:
raise JSONRPCValidationError(colander_exc=err)
try:
gist = GistModel().create(
owner=owner,
gist_id=schema_data['gistid'],
description=schema_data['description'],
gist_mapping=schema_data['nodes'],
gist_type=schema_data['gist_type'],
lifetime=schema_data['lifetime'],
gist_acl_level=schema_data['gist_acl_level'])
Session().commit()
return {
'msg': 'created new gist',

4
rhodecode/api/views/repo_api.py
View file

@ -44,7 +44,7 @@ from rhodecode.model.repo import RepoModel
from rhodecode.model.repo_group import RepoGroupModel
from rhodecode.model.scm import ScmModel, RepoList
from rhodecode.model.settings import SettingsModel, VcsSettingsModel
from rhodecode.model.validation_schema import RepoSchema
from rhodecode.model.validation_schema.schemas import repo_schema
log = logging.getLogger(__name__)
@ -610,7 +610,7 @@ def create_repo(request, apiuser, repo_name, repo_type,
}
"""
schema = RepoSchema()
schema = repo_schema.RepoSchema()
try:
data = schema.deserialize({
'repo_name': repo_name

4
rhodecode/api/views/repo_group_api.py
View file

@ -34,7 +34,7 @@ from rhodecode.lib.auth import (
from rhodecode.model.db import Session, RepoGroup
from rhodecode.model.repo_group import RepoGroupModel
from rhodecode.model.scm import RepoGroupList
from rhodecode.model.validation_schema import RepoGroupSchema
from rhodecode.model.validation_schema.schemas import repo_group_schema
log = logging.getLogger(__name__)
@ -193,7 +193,7 @@ def create_repo_group(request, apiuser, group_name, description=Optional(''),
"""
schema = RepoGroupSchema()
schema = repo_group_schema.RepoGroupSchema()
try:
data = schema.deserialize({
'group_name': group_name

130
rhodecode/controllers/admin/gists.py
View file

@ -25,15 +25,18 @@ gist controller for RhodeCode
import time
import logging
import traceback
import formencode
import peppercorn
from formencode import htmlfill
from pylons import request, response, tmpl_context as c, url
from pylons.controllers.util import abort, redirect
from pylons.i18n.translation import _
from webob.exc import HTTPNotFound, HTTPForbidden
from sqlalchemy.sql.expression import or_
from rhodecode.model.forms import GistForm
from rhodecode.model.gist import GistModel
from rhodecode.model.meta import Session
from rhodecode.model.db import Gist, User
@ -44,9 +47,10 @@ from rhodecode.lib.auth import LoginRequired, NotAnonymous
from rhodecode.lib.utils import jsonify
from rhodecode.lib.utils2 import safe_str, safe_int, time_to_datetime
from rhodecode.lib.ext_json import json
from webob.exc import HTTPNotFound, HTTPForbidden
from sqlalchemy.sql.expression import or_
from rhodecode.lib.vcs.exceptions import VCSError, NodeNotChangedError
from rhodecode.model import validation_schema
from rhodecode.model.validation_schema.schemas import gist_schema
log = logging.getLogger(__name__)
@ -56,11 +60,11 @@ class GistsController(BaseController):
def __load_defaults(self, extra_values=None):
c.lifetime_values = [
(str(-1), _('forever')),
(str(5), _('5 minutes')),
(str(60), _('1 hour')),
(str(60 * 24), _('1 day')),
(str(60 * 24 * 30), _('1 month')),
(-1, _('forever')),
(5, _('5 minutes')),
(60, _('1 hour')),
(60 * 24, _('1 day')),
(60 * 24 * 30, _('1 month')),
]
if extra_values:
c.lifetime_values.append(extra_values)
@ -136,40 +140,56 @@ class GistsController(BaseController):
"""POST /admin/gists: Create a new item"""
# url('gists')
self.__load_defaults()
gist_form = GistForm([x[0] for x in c.lifetime_values],
[x[0] for x in c.acl_options])()
data = dict(request.POST)
data['filename'] = data.get('filename') or Gist.DEFAULT_FILENAME
data['nodes'] = [{
'filename': data['filename'],
'content': data.get('content'),
'mimetype': data.get('mimetype') # None is autodetect
}]
data['gist_type'] = (
Gist.GIST_PUBLIC if data.get('public') else Gist.GIST_PRIVATE)
data['gist_acl_level'] = (
data.get('gist_acl_level') or Gist.ACL_LEVEL_PRIVATE)
schema = gist_schema.GistSchema().bind(
lifetime_options=[x[0] for x in c.lifetime_values])
try:
form_result = gist_form.to_python(dict(request.POST))
# TODO: multiple files support, from the form
filename = form_result['filename'] or Gist.DEFAULT_FILENAME
nodes = {
filename: {
'content': form_result['content'],
'lexer': form_result['mimetype'] # None is autodetect
}
}
_public = form_result['public']
gist_type = Gist.GIST_PUBLIC if _public else Gist.GIST_PRIVATE
gist_acl_level = form_result.get(
'acl_level', Gist.ACL_LEVEL_PRIVATE)
schema_data = schema.deserialize(data)
# convert to safer format with just KEYs so we sure no duplicates
schema_data['nodes'] = gist_schema.sequence_to_nodes(
schema_data['nodes'])
gist = GistModel().create(
description=form_result['description'],
gist_id=schema_data['gistid'], # custom access id not real ID
description=schema_data['description'],
owner=c.rhodecode_user.user_id,
gist_mapping=nodes,
gist_type=gist_type,
lifetime=form_result['lifetime'],
gist_id=form_result['gistid'],
gist_acl_level=gist_acl_level
gist_mapping=schema_data['nodes'],
gist_type=schema_data['gist_type'],
lifetime=schema_data['lifetime'],
gist_acl_level=schema_data['gist_acl_level']
)
Session().commit()
new_gist_id = gist.gist_access_id
except formencode.Invalid as errors:
defaults = errors.value
except validation_schema.Invalid as errors:
defaults = data
errors = errors.asdict()
if 'nodes.0.content' in errors:
errors['content'] = errors['nodes.0.content']
del errors['nodes.0.content']
if 'nodes.0.filename' in errors:
errors['filename'] = errors['nodes.0.filename']
del errors['nodes.0.filename']
return formencode.htmlfill.render(
render('admin/gists/new.html'),
defaults=defaults,
errors=errors.error_dict or {},
errors=errors,
prefix_error=False,
encoding="UTF-8",
force_defaults=False
@ -243,7 +263,8 @@ class GistsController(BaseController):
log.exception("Exception in gist show")
raise HTTPNotFound()
if format == 'raw':
content = '\n\n'.join([f.content for f in c.files if (f_path is None or f.path == f_path)])
content = '\n\n'.join([f.content for f in c.files
if (f_path is None or f.path == f_path)])
response.content_type = 'text/plain'
return content
return render('admin/gists/show.html')
@ -252,32 +273,35 @@ class GistsController(BaseController):
@NotAnonymous()
@auth.CSRFRequired()
def edit(self, gist_id):
self.__load_defaults()
self._add_gist_to_context(gist_id)
owner = c.gist.gist_owner == c.rhodecode_user.user_id
if not (h.HasPermissionAny('hg.admin')() or owner):
raise HTTPForbidden()
rpost = request.POST
nodes = {}
_file_data = zip(rpost.getall('org_files'), rpost.getall('files'),
rpost.getall('mimetypes'), rpost.getall('contents'))
for org_filename, filename, mimetype, content in _file_data:
nodes[org_filename] = {
'org_filename': org_filename,
'filename': filename,
'content': content,
'lexer': mimetype,
}
data = peppercorn.parse(request.POST.items())
schema = gist_schema.GistSchema()
schema = schema.bind(
# '0' is special value to leave lifetime untouched
lifetime_options=[x[0] for x in c.lifetime_values] + [0],
)
try:
schema_data = schema.deserialize(data)
# convert to safer format with just KEYs so we sure no duplicates
schema_data['nodes'] = gist_schema.sequence_to_nodes(
schema_data['nodes'])
GistModel().update(
gist=c.gist,
description=rpost['description'],
description=schema_data['description'],
owner=c.gist.owner,
gist_mapping=nodes,
gist_type=c.gist.gist_type,
lifetime=rpost['lifetime'],
gist_acl_level=rpost['acl_level']
gist_mapping=schema_data['nodes'],
gist_type=schema_data['gist_type'],
lifetime=schema_data['lifetime'],
gist_acl_level=schema_data['gist_acl_level']
)
Session().commit()
@ -287,6 +311,10 @@ class GistsController(BaseController):
# store only DB stuff for gist
Session().commit()
h.flash(_('Successfully updated gist data'), category='success')
except validation_schema.Invalid as errors:
errors = errors.asdict()
h.flash(_('Error occurred during update of gist {}: {}').format(
gist_id, errors), category='error')
except Exception:
log.exception("Exception in gist edit")
h.flash(_('Error occurred during update of gist %s') % gist_id,
@ -317,7 +345,7 @@ class GistsController(BaseController):
# this cannot use timeago, since it's used in select2 as a value
expiry = h.age(h.time_to_datetime(c.gist.gist_expires))
self.__load_defaults(
extra_values=('0', _('%(expiry)s - current value') % {'expiry': expiry}))
extra_values=(0, _('%(expiry)s - current value') % {'expiry': expiry}))
return render('admin/gists/edit.html')
@LoginRequired()

4
rhodecode/controllers/search.py
View file

@ -35,6 +35,7 @@ from rhodecode.lib.helpers import Page
from rhodecode.lib.utils2 import safe_str, safe_int
from rhodecode.lib.index import searcher_from_config
from rhodecode.model import validation_schema
from rhodecode.model.validation_schema.schemas import search_schema
log = logging.getLogger(__name__)
@ -48,7 +49,7 @@ class SearchController(BaseRepoController):
formatted_results = []
execution_time = ''
schema = validation_schema.SearchParamsSchema()
schema = search_schema.SearchParamsSchema()
search_params = {}
errors = []
@ -75,7 +76,6 @@ class SearchController(BaseRepoController):
page_limit = search_params['page_limit']
requested_page = search_params['requested_page']
c.perm_user = AuthUser(user_id=c.rhodecode_user.user_id,
ip_addr=self.ip_addr)

17
rhodecode/model/forms.py
View file

@ -555,23 +555,6 @@ def PullRequestForm(repo_id):
return _PullRequestForm
def GistForm(lifetime_options, acl_level_options):
class _GistForm(formencode.Schema):
gistid = All(v.UniqGistId(), v.UnicodeString(strip=True, min=3, not_empty=False, if_missing=None))
filename = All(v.BasePath()(),
v.UnicodeString(strip=True, required=False))
description = v.UnicodeString(required=False, if_missing=u'')
lifetime = v.OneOf(lifetime_options)
mimetype = v.UnicodeString(required=False, if_missing=None)
content = v.UnicodeString(required=True, not_empty=True)
public = v.UnicodeString(required=False, if_missing=u'')
private = v.UnicodeString(required=False, if_missing=u'')
acl_level = v.OneOf(acl_level_options)
return _GistForm
def IssueTrackerPatternsForm():
class _IssueTrackerPatternsForm(formencode.Schema):
allow_extra_fields = True

28
rhodecode/model/gist.py
View file

@ -107,7 +107,7 @@ class GistModel(BaseModel):
:param description: description of the gist
:param owner: user who created this gist
:param gist_mapping: mapping {filename:{'content':content},...}
:param gist_mapping: mapping [{'filename': 'file1.txt', 'content': content}, ...}]
:param gist_type: type of gist private/public
:param lifetime: in minutes, -1 == forever
:param gist_acl_level: acl level for this gist
@ -141,25 +141,10 @@ class GistModel(BaseModel):
repo_name=gist_id, repo_type='hg', repo_group=GIST_STORE_LOC,
use_global_config=True)
processed_mapping = {}
for filename in gist_mapping:
if filename != os.path.basename(filename):
raise Exception('Filename cannot be inside a directory')
content = gist_mapping[filename]['content']
# TODO: expand support for setting explicit lexers
# if lexer is None:
# try:
# guess_lexer = pygments.lexers.guess_lexer_for_filename
# lexer = guess_lexer(filename,content)
# except pygments.util.ClassNotFound:
# lexer = 'text'
processed_mapping[filename] = {'content': content}
# now create single multifile commit
message = 'added file'
message += 's: ' if len(processed_mapping) > 1 else ': '
message += ', '.join([x for x in processed_mapping])
message += 's: ' if len(gist_mapping) > 1 else ': '
message += ', '.join([x for x in gist_mapping])
# fake RhodeCode Repository object
fake_repo = AttributeDict({
@ -170,7 +155,7 @@ class GistModel(BaseModel):
ScmModel().create_nodes(
user=owner.user_id, repo=fake_repo,
message=message,
nodes=processed_mapping,
nodes=gist_mapping,
trigger_push_hook=False
)
@ -196,7 +181,6 @@ class GistModel(BaseModel):
gist = self._get_gist(gist)
gist_repo = gist.scm_instance()
lifetime = safe_int(lifetime, -1)
if lifetime == 0: # preserve old value
gist_expires = gist.gist_expires
else:
@ -207,9 +191,9 @@ class GistModel(BaseModel):
gist_mapping_op = {}
for k, v in gist_mapping.items():
# add, mod, del
if not v['org_filename'] and v['filename']:
if not v['filename_org'] and v['filename']:
op = 'add'
elif v['org_filename'] and not v['filename']:
elif v['filename_org'] and not v['filename']:
op = 'del'
else:
op = 'mod'

24
rhodecode/model/validation_schema/__init__.py Normal file
View file

@ -0,0 +1,24 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016-2016 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# 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 Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import colander
from colander import Invalid # noqa, don't remove this

89
rhodecode/model/validation_schema/preparers.py Normal file
View file

@ -0,0 +1,89 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016-2016 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# 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 Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import unicodedata
def strip_preparer(value):
"""
strips given values using .strip() function
"""
if value:
value = value.strip()
return value
def slugify_preparer(value):
"""
Slugify given value to a safe representation for url/id
"""
from rhodecode.lib.utils import repo_name_slug
if value:
value = repo_name_slug(value.lower())
return value
def non_ascii_strip_preparer(value):
"""
trie to replace non-ascii letters to their ascii representation
eg::
`żołw` converts into `zolw`
"""
if value:
value = unicodedata.normalize('NFKD', value).encode('ascii', 'ignore')
return value
def unique_list_preparer(value):
"""
Converts an list to a list with only unique values
"""
def make_unique(value):
seen = []
return [c for c in value if
not (c in seen or seen.append(c))]
if isinstance(value, list):
ret_val = make_unique(value)
elif isinstance(value, set):
ret_val = list(value)
elif isinstance(value, tuple):
ret_val = make_unique(value)
elif value is None:
ret_val = []
else:
ret_val = [value]
return ret_val
def unique_list_from_str_preparer(value):
"""
Converts an list to a list with only unique values
"""
from rhodecode.lib.utils2 import aslist
if isinstance(value, basestring):
value = aslist(value, ',')
return unique_list_preparer(value)

25
rhodecode/model/validation_schema/schemas/__init__.py Normal file
View file

@ -0,0 +1,25 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016-2016 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# 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 Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
"""
Colander Schema nodes
http://docs.pylonsproject.org/projects/colander/en/latest/basics.html#schema-node-objects
"""

185
rhodecode/model/validation_schema/schemas/gist_schema.py Normal file
View file

@ -0,0 +1,185 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016-2016 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# 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 Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import os
import colander
from rhodecode.translation import _
from rhodecode.model.validation_schema import validators, preparers
def nodes_to_sequence(nodes, colander_node=None):
"""
Converts old style dict nodes to new list of dicts
:param nodes: dict with key beeing name of the file
"""
if not isinstance(nodes, dict):
msg = 'Nodes needs to be a dict, got {}'.format(type(nodes))
raise colander.Invalid(colander_node, msg)
out = []
for key, val in nodes.items():
val = (isinstance(val, dict) and val) or {}
out.append(dict(
filename=key,
content=val.get('content'),
mimetype=val.get('mimetype')
))
out = Nodes().deserialize(out)
return out
def sequence_to_nodes(nodes, colander_node=None):
if not isinstance(nodes, list):
msg = 'Nodes needs to be a list, got {}'.format(type(nodes))
raise colander.Invalid(colander_node, msg)
nodes = Nodes().deserialize(nodes)
out = {}
try:
for file_data in nodes:
file_data_skip = file_data.copy()
# if we got filename_org we use it as a key so we keep old
# name as input and rename is-reflected inside the values as
# filename and filename_org differences.
filename_org = file_data.get('filename_org')
filename = filename_org or file_data['filename']
out[filename] = {}
out[filename].update(file_data_skip)
except Exception as e:
msg = 'Invalid data format org_exc:`{}`'.format(repr(e))
raise colander.Invalid(colander_node, msg)
return out
@colander.deferred
def deferred_lifetime_validator(node, kw):
options = kw.get('lifetime_options', [])
return colander.All(
colander.Range(min=-1, max=60 * 24 * 30 * 12),
colander.OneOf([x for x in options]))
def unique_gist_validator(node, value):
from rhodecode.model.db import Gist
existing = Gist.get_by_access_id(value)
if existing:
msg = _(u'Gist with name {} already exists').format(value)
raise colander.Invalid(node, msg)
def filename_validator(node, value):
if value != os.path.basename(value):
msg = _(u'Filename {} cannot be inside a directory').format(value)
raise colander.Invalid(node, msg)
class NodeSchema(colander.MappingSchema):
# if we perform rename this will be org filename
filename_org = colander.SchemaNode(
colander.String(),
preparer=[preparers.strip_preparer,
preparers.non_ascii_strip_preparer],
validator=filename_validator,
missing=None)
filename = colander.SchemaNode(
colander.String(),
preparer=[preparers.strip_preparer,
preparers.non_ascii_strip_preparer],
validator=filename_validator)
content = colander.SchemaNode(
colander.String())
mimetype = colander.SchemaNode(
colander.String(),
missing=None)
class Nodes(colander.SequenceSchema):
filenames = NodeSchema()
def validator(self, node, cstruct):
if not isinstance(cstruct, list):
return
found_filenames = []
for data in cstruct:
filename = data['filename']
if filename in found_filenames:
msg = _('Duplicated value for filename found: `{}`').format(
filename)
raise colander.Invalid(node, msg)
found_filenames.append(filename)
class GistSchema(colander.MappingSchema):
"""
schema = GistSchema()
schema.bind(
lifetime_options = [1,2,3]
)
out = schema.deserialize(dict(
nodes=[
{'filename': 'x', 'content': 'xxx', },
{'filename': 'docs/Z', 'content': 'xxx', 'mimetype': 'x'},
]
))
"""
from rhodecode.model.db import Gist
gistid = colander.SchemaNode(
colander.String(),
missing=None,
preparer=[preparers.strip_preparer,
preparers.non_ascii_strip_preparer,
preparers.slugify_preparer],
validator=colander.All(
colander.Length(min=3),
unique_gist_validator
))
description = colander.SchemaNode(
colander.String(),
missing=u'')
lifetime = colander.SchemaNode(
colander.Integer(),
validator=deferred_lifetime_validator)
gist_acl_level = colander.SchemaNode(
colander.String(),
validator=colander.OneOf([Gist.ACL_LEVEL_PUBLIC,
Gist.ACL_LEVEL_PRIVATE]))
gist_type = colander.SchemaNode(
colander.String(),
missing=Gist.ACL_LEVEL_PUBLIC,
validator=colander.OneOf([Gist.GIST_PRIVATE, Gist.GIST_PUBLIC]))
nodes = Nodes()

29
rhodecode/model/validation_schema/schemas/repo_group_schema.py Normal file
View file

@ -0,0 +1,29 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016-2016 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# 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 Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import colander
from rhodecode.model.validation_schema import validators, preparers, types
class RepoGroupSchema(colander.Schema):
group_name = colander.SchemaNode(types.GroupNameType())

27
rhodecode/model/validation_schema/schemas/repo_schema.py Normal file
View file

@ -0,0 +1,27 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016-2016 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# 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 Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import colander
from rhodecode.model.validation_schema import validators, preparers, types
class RepoSchema(colander.Schema):
repo_name = colander.SchemaNode(types.GroupNameType())

24
rhodecode/model/validation_schema.py → rhodecode/model/validation_schema/schemas/search_schema.py
View file

@ -18,29 +18,8 @@
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import colander
from colander import Invalid # noqa
class GroupNameType(colander.String):
SEPARATOR = '/'
def deserialize(self, node, cstruct):
result = super(GroupNameType, self).deserialize(node, cstruct)
return self._replace_extra_slashes(result)
def _replace_extra_slashes(self, path):
path = path.split(self.SEPARATOR)
path = [item for item in path if item]
return self.SEPARATOR.join(path)
class RepoGroupSchema(colander.Schema):
group_name = colander.SchemaNode(GroupNameType())
class RepoSchema(colander.Schema):
repo_name = colander.SchemaNode(GroupNameType())
class SearchParamsSchema(colander.MappingSchema):
@ -63,4 +42,3 @@ class SearchParamsSchema(colander.MappingSchema):
requested_page = colander.SchemaNode(
colander.Integer(),
missing=1)

34
rhodecode/model/validation_schema/types.py Normal file
View file

@ -0,0 +1,34 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016-2016 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# 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 Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import colander
class GroupNameType(colander.String):
SEPARATOR = '/'
def deserialize(self, node, cstruct):
result = super(GroupNameType, self).deserialize(node, cstruct)
return self._replace_extra_slashes(result)
def _replace_extra_slashes(self, path):
path = path.split(self.SEPARATOR)
path = [item for item in path if item]
return self.SEPARATOR.join(path)

19
rhodecode/model/validation_schema/validators.py Normal file
View file

@ -0,0 +1,19 @@
import os
import ipaddress
import colander
from rhodecode.translation import _
def ip_addr_validator(node, value):
try:
# this raises an ValueError if address is not IpV4 or IpV6
ipaddress.ip_network(value, strict=False)
except ValueError:
msg = _(u'Please enter a valid IPv4 or IpV6 address')
raise colander.Invalid(node, msg)

36
rhodecode/model/validators.py
View file

@ -970,22 +970,6 @@ def FieldKey():
return _validator
def BasePath():
class _validator(formencode.validators.FancyValidator):
messages = {
'badPath': _(u'Filename cannot be inside a directory'),
}
def _to_python(self, value, state):
return value
def validate_python(self, value, state):
if value != os.path.basename(value):
raise formencode.Invalid(self.message('badPath', state),
value, state)
return _validator
def ValidAuthPlugins():
class _validator(formencode.validators.FancyValidator):
messages = {
@ -1061,26 +1045,6 @@ def ValidAuthPlugins():
return _validator
def UniqGistId():
class _validator(formencode.validators.FancyValidator):
messages = {
'gistid_taken': _(u'This gistid is already in use')
}
def _to_python(self, value, state):
return repo_name_slug(value.lower())
def validate_python(self, value, state):
existing = Gist.get_by_access_id(value)
if existing:
msg = M(self, 'gistid_taken', state)
raise formencode.Invalid(
msg, value, state, error_dict={'gistid': msg}
)
return _validator
def ValidPattern():
class _Validator(formencode.validators.FancyValidator):

22
rhodecode/templates/admin/gists/edit.html
View file

@ -44,27 +44,31 @@
<label for='lifetime'>${_('Gist lifetime')}</label>
${h.dropdownmenu('lifetime', '0', c.lifetime_options)}
<label for='acl_level'>${_('Gist access level')}</label>
${h.dropdownmenu('acl_level', c.gist.acl_level, c.acl_options)}
<label for='gist_acl_level'>${_('Gist access level')}</label>
${h.dropdownmenu('gist_acl_level', c.gist.acl_level, c.acl_options)}
</div>
</div>
## peppercorn schema
<input type="hidden" name="__start__" value="nodes:sequence"/>
% for cnt, file in enumerate(c.files):
<input type="hidden" name="__start__" value="file:mapping"/>
<div id="codeblock" class="codeblock" >
<div class="code-header">
<div class="form">
<div class="fields">
<input type="hidden" value="${file.path}" name="org_files">
<input id="filename_${h.FID('f',file.path)}" name="files" size="30" type="text" value="${file.path}">
${h.dropdownmenu('mimetypes' ,'plain',[('plain',_('plain'))],enable_filter=True, id='mimetype_'+h.FID('f',file.path))}
<input type="hidden" name="filename_org" value="${file.path}" >
<input id="filename_${h.FID('f',file.path)}" name="filename" size="30" type="text" value="${file.path}">
${h.dropdownmenu('mimetype' ,'plain',[('plain',_('plain'))],enable_filter=True, id='mimetype_'+h.FID('f',file.path))}
</div>
</div>
</div>
<div class="editor_container">
<pre id="editor_pre"></pre>
<textarea id="editor_${h.FID('f',file.path)}" name="contents" >${file.content}</textarea>
<textarea id="editor_${h.FID('f',file.path)}" name="content" >${file.content}</textarea>
</div>
</div>
<input type="hidden" name="__end__" />
## dynamic edit box.
<script type="text/javascript">
@ -72,7 +76,7 @@
var myCodeMirror = initCodeMirror(
"editor_${h.FID('f',file.path)}", '');
var modes_select = $('#mimetype_${h.FID('f',file.path)}');
var modes_select = $("#mimetype_${h.FID('f',file.path)}");
fillCodeMirrorOptions(modes_select);
// try to detect the mode based on the file we edit
@ -86,7 +90,7 @@
setCodeMirrorMode(myCodeMirror, detected_mode);
}
var filename_selector = '#filename_${h.FID('f',file.path)}';
var filename_selector = "#filename_${h.FID('f',file.path)}";
// on change of select field set mode
setCodeMirrorModeFromSelect(
modes_select, filename_selector, myCodeMirror, null);
@ -96,8 +100,8 @@
modes_select, filename_selector, myCodeMirror, null);
});
</script>
%endfor
<input type="hidden" name="__end__" />
<div class="pull-right">
${h.submit('update',_('Update Gist'),class_="btn btn-success")}

2
rhodecode/templates/admin/gists/new.html
View file

@ -39,7 +39,7 @@
${h.dropdownmenu('lifetime', '', c.lifetime_options)}
<label for='acl_level'>${_('Gist access level')}</label>
${h.dropdownmenu('acl_level', '', c.acl_options)}
${h.dropdownmenu('gist_acl_level', '', c.acl_options)}
</div>
<div id="codeblock" class="codeblock">

98
rhodecode/tests/functional/test_admin_gists.py
View file

@ -129,40 +129,33 @@ class TestGistsController(TestController):
for gist in GistModel.get_all():
response.mustcontain(no=['gist: %s' % gist.gist_access_id])
def test_create_missing_description(self):
def test_create(self):
self.log_user()
response = self.app.post(
url('gists'),
params={'lifetime': -1, 'csrf_token': self.csrf_token},
status=200)
response.mustcontain('Missing value')
def test_create(self):
self.log_user()
response = self.app.post(url('gists'),
params={'lifetime': -1,
'content': 'gist test',
'filename': 'foo',
'public': 'public',
'acl_level': Gist.ACL_LEVEL_PUBLIC,
'csrf_token': self.csrf_token},
status=302)
params={'lifetime': -1,
'content': 'gist test',
'filename': 'foo',
'public': 'public',
'gist_acl_level': Gist.ACL_LEVEL_PUBLIC,
'csrf_token': self.csrf_token},
status=302)
response = response.follow()
response.mustcontain('added file: foo')
response.mustcontain('gist test')
def test_create_with_path_with_dirs(self):
self.log_user()
response = self.app.post(url('gists'),
params={'lifetime': -1,
'content': 'gist test',
'filename': '/home/foo',
'public': 'public',
'acl_level': Gist.ACL_LEVEL_PUBLIC,
'csrf_token': self.csrf_token},
status=200)
response.mustcontain('Filename cannot be inside a directory')
response = self.app.post(
url('gists'),
params={'lifetime': -1,
'content': 'gist test',
'filename': '/home/foo',
'public': 'public',
'gist_acl_level': Gist.ACL_LEVEL_PUBLIC,
'csrf_token': self.csrf_token},
status=200)
response.mustcontain('Filename /home/foo cannot be inside a directory')
def test_access_expired_gist(self, create_gist):
self.log_user()
@ -175,14 +168,15 @@ class TestGistsController(TestController):
def test_create_private(self):
self.log_user()
response = self.app.post(url('gists'),
params={'lifetime': -1,
'content': 'private gist test',
'filename': 'private-foo',
'private': 'private',
'acl_level': Gist.ACL_LEVEL_PUBLIC,
'csrf_token': self.csrf_token},
status=302)
response = self.app.post(
url('gists'),
params={'lifetime': -1,
'content': 'private gist test',
'filename': 'private-foo',
'private': 'private',
'gist_acl_level': Gist.ACL_LEVEL_PUBLIC,
'csrf_token': self.csrf_token},
status=302)
response = response.follow()
response.mustcontain('added file: private-foo<')
response.mustcontain('private gist test')
@ -193,14 +187,15 @@ class TestGistsController(TestController):
def test_create_private_acl_private(self):
self.log_user()
response = self.app.post(url('gists'),
params={'lifetime': -1,
'content': 'private gist test',
'filename': 'private-foo',
'private': 'private',
'acl_level': Gist.ACL_LEVEL_PRIVATE,
'csrf_token': self.csrf_token},
status=302)
response = self.app.post(
url('gists'),
params={'lifetime': -1,
'content': 'private gist test',
'filename': 'private-foo',
'private': 'private',
'gist_acl_level': Gist.ACL_LEVEL_PRIVATE,
'csrf_token': self.csrf_token},
status=302)
response = response.follow()
response.mustcontain('added file: private-foo<')
response.mustcontain('private gist test')
@ -211,15 +206,16 @@ class TestGistsController(TestController):
def test_create_with_description(self):
self.log_user()
response = self.app.post(url('gists'),
params={'lifetime': -1,
'content': 'gist test',
'filename': 'foo-desc',
'description': 'gist-desc',
'public': 'public',
'acl_level': Gist.ACL_LEVEL_PUBLIC,
'csrf_token': self.csrf_token},
status=302)
response = self.app.post(
url('gists'),
params={'lifetime': -1,
'content': 'gist test',
'filename': 'foo-desc',
'description': 'gist-desc',
'public': 'public',
'gist_acl_level': Gist.ACL_LEVEL_PUBLIC,
'csrf_token': self.csrf_token},
status=302)
response = response.follow()
response.mustcontain('added file: foo-desc')
response.mustcontain('gist test')
@ -233,7 +229,7 @@ class TestGistsController(TestController):
'filename': 'foo-desc',
'description': 'gist-desc',
'public': 'public',
'acl_level': Gist.ACL_LEVEL_PUBLIC,
'gist_acl_level': Gist.ACL_LEVEL_PUBLIC,
'csrf_token': self.csrf_token
}
response = self.app.post(url('gists'), params=params, status=302)

0
rhodecode/tests/models/schemas/__init__.py Normal file
View file

100
rhodecode/tests/models/schemas/test_gist_schema.py Normal file
View file

@ -0,0 +1,100 @@
# -*- coding: utf-8 -*-
# Copyright (C) 2016-2016 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# 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 Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import colander
import pytest
from rhodecode.model import validation_schema
from rhodecode.model.validation_schema.schemas import gist_schema
class TestGistSchema(object):
def test_deserialize_bad_data(self):
schema = gist_schema.GistSchema().bind(
lifetime_options=[1, 2, 3]
)
with pytest.raises(validation_schema.Invalid) as exc_info:
schema.deserialize('err')
err = exc_info.value.asdict()
assert err[''] == '"err" is not a mapping type: ' \
'Does not implement dict-like functionality.'
def test_deserialize_bad_lifetime_options(self):
schema = gist_schema.GistSchema().bind(
lifetime_options=[1, 2, 3]
)
with pytest.raises(validation_schema.Invalid) as exc_info:
schema.deserialize(dict(
lifetime=10
))
err = exc_info.value.asdict()
assert err['lifetime'] == '"10" is not one of 1, 2, 3'
with pytest.raises(validation_schema.Invalid) as exc_info:
schema.deserialize(dict(
lifetime='x'
))
err = exc_info.value.asdict()
assert err['lifetime'] == '"x" is not a number'
def test_serialize_data_correctly(self):
schema = gist_schema.GistSchema().bind(
lifetime_options=[1, 2, 3]
)
nodes = [{
'filename': 'foobar',
'filename_org': 'foobar',
'content': 'content',
'mimetype': 'xx'
}]
schema_data = schema.deserialize(dict(
lifetime=2,
gist_type='public',
gist_acl_level='acl_public',
nodes=nodes,
))
assert schema_data['nodes'] == nodes
def test_serialize_data_correctly_with_conversion(self):
schema = gist_schema.GistSchema().bind(
lifetime_options=[1, 2, 3],
convert_nodes=True
)
nodes = [{
'filename': 'foobar',
'filename_org': None,
'content': 'content',
'mimetype': 'xx'
}]
schema_data = schema.deserialize(dict(
lifetime=2,
gist_type='public',
gist_acl_level='acl_public',
nodes=nodes,
))
assert schema_data['nodes'] == nodes
seq_nodes = gist_schema.sequence_to_nodes(nodes)
assert isinstance(seq_nodes, dict)
seq_nodes = gist_schema.nodes_to_sequence(seq_nodes)
assert nodes == seq_nodes

2
rhodecode/tests/models/test_validation_schema.py → rhodecode/tests/models/schemas/test_schema_types.py
View file

@ -21,7 +21,7 @@
import colander
import pytest
from rhodecode.model.validation_schema import GroupNameType
from rhodecode.model.validation_schema.types import GroupNameType
class TestGroupNameType(object):