FileController.get_one() - Code Metrics - Inspection of "Merge pull request #3300 from StackStorm/fix/kwarg..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — develop ( a82689...1ba140 )

by Plexxi

created 2017-03-25 10:00 UTC

FileController.get_one() C

↳ Parent: FileController

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	9
dl	0
loc	55
rs	5.4159
c	0
b	0
f	0

How to fix Long Method

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import codecs
import mimetypes
import os

import six
from wsgiref.handlers import format_date_time

from st2api.controllers.v1.packs import BasePacksController
from st2common.exceptions.db import StackStormDBObjectNotFoundError
from st2common import log as logging
from st2common.models.api.pack import PackAPI
from st2common.persistence.pack import Pack
from st2common.content.utils import get_pack_file_abs_path
from st2common.rbac.types import PermissionType
from st2common.rbac import utils as rbac_utils
from st2common.router import abort
from st2common.router import Response

http_client = six.moves.http_client

__all__ = [
    'FilesController',
    'FileController'
]

http_client = six.moves.http_client

LOG = logging.getLogger(__name__)

BOM_LEN = len(codecs.BOM_UTF8)

# Maximum file size in bytes. If the file on disk is larger then this value, we don't include it
# in the response. This prevents DDoS / exhaustion attacks.
MAX_FILE_SIZE = (500 * 1000)

# File paths in the file controller for which RBAC checks are not performed
WHITELISTED_FILE_PATHS = [
    'icon.png'
]


class BaseFileController(BasePacksController):
    model = PackAPI
    access = Pack

    supported_filters = {}
    query_options = {}

    def get_all(self):

        return abort(404)

    def _get_file_size(self, file_path):
        return self._get_file_stats(file_path=file_path)[0]

    def _get_file_stats(self, file_path):
        try:
            file_stats = os.stat(file_path)
        except OSError:
            return (None, None)

        return file_stats.st_size, file_stats.st_mtime

    def _get_file_content(self, file_path):
        with codecs.open(file_path, 'rb') as fp:
            content = fp.read()

        return content

    def _process_file_content(self, content):
        """
        This method processes the file content and removes unicode BOM character if one is present.

        Note: If we don't do that, files view explodes with "UnicodeDecodeError: ... invalid start
        byte" because the json.dump doesn't know how to handle BOM character.
        """
        if content.startswith(codecs.BOM_UTF8):
            content = content[BOM_LEN:]

        return content


class FilesController(BaseFileController):
    """
    Controller which allows user to retrieve content of all the files inside the pack.
    """

    def __init__(self):
        super(FilesController, self).__init__()
        self.get_one_db_method = self._get_by_ref_or_id

    def get_one(self, ref_or_id, requester_user):

        """
            Outputs the content of all the files inside the pack.

            Handles requests:
                GET /packs/views/files/<pack_ref_or_id>
        """
        pack_db = self._get_by_ref_or_id(ref_or_id=ref_or_id)

        rbac_utils.assert_user_has_resource_db_permission(user_db=requester_user,
                                                          resource_db=pack_db,
                                                          permission_type=PermissionType.PACK_VIEW)

        if not pack_db:
            msg = 'Pack with ref_or_id "%s" does not exist' % (ref_or_id)
            raise StackStormDBObjectNotFoundError(msg)

        pack_ref = pack_db.ref
        pack_files = pack_db.files

        result = []
        for file_path in pack_files:
            normalized_file_path = get_pack_file_abs_path(pack_ref=pack_ref, file_path=file_path)
            if not normalized_file_path or not os.path.isfile(normalized_file_path):
                # Ignore references to files which don't exist on disk
                continue

            file_size = self._get_file_size(file_path=normalized_file_path)
            if file_size is not None and file_size > MAX_FILE_SIZE:
                LOG.debug('Skipping file "%s" which size exceeds max file size (%s bytes)' %
                          (normalized_file_path, MAX_FILE_SIZE))
                continue

            content = self._get_file_content(file_path=normalized_file_path)

            include_file = self._include_file(file_path=file_path, content=content)
            if not include_file:
                LOG.debug('Skipping binary file "%s"' % (normalized_file_path))
                continue

            item = {
                'file_path': file_path,
                'content': content
            }
            result.append(item)
        return result

    def _include_file(self, file_path, content):
        """
        Method which returns True if the following file content should be included in the response.

        Right now we exclude any file with UTF8 BOM character in it - those are most likely binary
        files such as icon, etc.
        """
        if codecs.BOM_UTF8 in content[:1024]:
            return False

        if "\0" in content[:1024]:
            # Found null byte, most likely a binary file
            return False

        return True


class FileController(BaseFileController):
    """
    Controller which allows user to retrieve content of a specific file in a pack.
    """

    def get_one(self, ref_or_id, file_path, requester_user, if_none_match=None,

                if_modified_since=None):
        """
            Outputs the content of a specific file in a pack.

            Handles requests:
                GET /packs/views/file/<pack_ref_or_id>/<file path>
        """
        pack_db = self._get_by_ref_or_id(ref_or_id=ref_or_id)

        if not pack_db:
            msg = 'Pack with ref_or_id "%s" does not exist' % (ref_or_id)
            raise StackStormDBObjectNotFoundError(msg)

        if not file_path:
            raise ValueError('Missing file path')

        pack_ref = pack_db.ref

        # Note: Until list filtering is in place we don't require RBAC check for icon file
        permission_type = PermissionType.PACK_VIEW
        if file_path not in WHITELISTED_FILE_PATHS:
            rbac_utils.assert_user_has_resource_db_permission(user_db=requester_user,
                                                              resource_db=pack_db,
                                                              permission_type=permission_type)

        normalized_file_path = get_pack_file_abs_path(pack_ref=pack_ref, file_path=file_path)
        if not normalized_file_path or not os.path.isfile(normalized_file_path):
            # Ignore references to files which don't exist on disk
            raise StackStormDBObjectNotFoundError('File "%s" not found' % (file_path))

        file_size, file_mtime = self._get_file_stats(file_path=normalized_file_path)

        response = Response()

        if not self._is_file_changed(file_mtime,
                                     if_none_match=if_none_match,
                                     if_modified_since=if_modified_since):
            response.status = http_client.NOT_MODIFIED
        else:
            if file_size is not None and file_size > MAX_FILE_SIZE:
                msg = ('File %s exceeds maximum allowed file size (%s bytes)' %
                       (file_path, MAX_FILE_SIZE))
                raise ValueError(msg)

            content_type = mimetypes.guess_type(normalized_file_path)[0] or \
                'application/octet-stream'

            response.headers['Content-Type'] = content_type
            response.body = self._get_file_content(file_path=normalized_file_path)

        response.headers['Last-Modified'] = format_date_time(file_mtime)
        response.headers['ETag'] = repr(file_mtime)

        return response

    def _is_file_changed(self, file_mtime, if_none_match=None, if_modified_since=None):
        # For if_none_match check against what would be the ETAG value
        if if_none_match:
            return repr(file_mtime) != if_none_match

        # For if_modified_since check against file_mtime
        if if_modified_since:
            return if_modified_since != format_date_time(file_mtime)

        # Neither header is provided therefore assume file is changed.
        return True


class PackViewsController(object):
    files = FilesController()
    file = FileController()


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			import codecs
17			import mimetypes
18			import os
19
20			import six
21			from wsgiref.handlers import format_date_time
22
23			from st2api.controllers.v1.packs import BasePacksController
24			from st2common.exceptions.db import StackStormDBObjectNotFoundError
25			from st2common import log as logging
26			from st2common.models.api.pack import PackAPI
27			from st2common.persistence.pack import Pack
28			from st2common.content.utils import get_pack_file_abs_path
29			from st2common.rbac.types import PermissionType
30			from st2common.rbac import utils as rbac_utils
31			from st2common.router import abort
32			from st2common.router import Response
33
34			http_client = six.moves.http_client
35
36			__all__ = [
37			'FilesController',
38			'FileController'
39			]
40
41			http_client = six.moves.http_client
42
43			LOG = logging.getLogger(__name__)
44
45			BOM_LEN = len(codecs.BOM_UTF8)
46
47			# Maximum file size in bytes. If the file on disk is larger then this value, we don't include it
48			# in the response. This prevents DDoS / exhaustion attacks.
49			MAX_FILE_SIZE = (500 * 1000)
50
51			# File paths in the file controller for which RBAC checks are not performed
52			WHITELISTED_FILE_PATHS = [
53			'icon.png'
54			]
55
56
57			class BaseFileController(BasePacksController):
58			model = PackAPI
59			access = Pack
60
61			supported_filters = {}
62			query_options = {}
63
64			def get_all(self):
			0 ignored issues – show Bug introduced 2017-03-24 05:41 UTC by Report Bug Copy Issue Report Arguments number differs from overridden 'get_all' method Loading history...
65			return abort(404)
66
67			def _get_file_size(self, file_path):
68			return self._get_file_stats(file_path=file_path)[0]
69
70			def _get_file_stats(self, file_path):
71			try:
72			file_stats = os.stat(file_path)
73			except OSError:
74			return (None, None)
75
76			return file_stats.st_size, file_stats.st_mtime
77
78			def _get_file_content(self, file_path):
79			with codecs.open(file_path, 'rb') as fp:
80			content = fp.read()
81
82			return content
83
84			def _process_file_content(self, content):
85			"""
86			This method processes the file content and removes unicode BOM character if one is present.
87
88			Note: If we don't do that, files view explodes with "UnicodeDecodeError: ... invalid start
89			byte" because the json.dump doesn't know how to handle BOM character.
90			"""
91			if content.startswith(codecs.BOM_UTF8):
92			content = content[BOM_LEN:]
93
94			return content
95
96
97			class FilesController(BaseFileController):
98			"""
99			Controller which allows user to retrieve content of all the files inside the pack.
100			"""
101
102			def __init__(self):
103			super(FilesController, self).__init__()
104			self.get_one_db_method = self._get_by_ref_or_id
105
106			def get_one(self, ref_or_id, requester_user):
			0 ignored issues – show Bug introduced 2017-02-13 12:24 UTC by Report Bug Copy Issue Report Arguments number differs from overridden 'get_one' method Loading history...
107			"""
108			Outputs the content of all the files inside the pack.
109
110			Handles requests:
111			GET /packs/views/files/<pack_ref_or_id>
112			"""
113			pack_db = self._get_by_ref_or_id(ref_or_id=ref_or_id)
114
115			rbac_utils.assert_user_has_resource_db_permission(user_db=requester_user,
116			resource_db=pack_db,
117			permission_type=PermissionType.PACK_VIEW)
118
119			if not pack_db:
120			msg = 'Pack with ref_or_id "%s" does not exist' % (ref_or_id)
121			raise StackStormDBObjectNotFoundError(msg)
122
123			pack_ref = pack_db.ref
124			pack_files = pack_db.files
125
126			result = []
127			for file_path in pack_files:
128			normalized_file_path = get_pack_file_abs_path(pack_ref=pack_ref, file_path=file_path)
129			if not normalized_file_path or not os.path.isfile(normalized_file_path):
130			# Ignore references to files which don't exist on disk
131			continue
132
133			file_size = self._get_file_size(file_path=normalized_file_path)
134			if file_size is not None and file_size > MAX_FILE_SIZE:
135			LOG.debug('Skipping file "%s" which size exceeds max file size (%s bytes)' %
136			(normalized_file_path, MAX_FILE_SIZE))
137			continue
138
139			content = self._get_file_content(file_path=normalized_file_path)
140
141			include_file = self._include_file(file_path=file_path, content=content)
142			if not include_file:
143			LOG.debug('Skipping binary file "%s"' % (normalized_file_path))
144			continue
145
146			item = {
147			'file_path': file_path,
148			'content': content
149			}
150			result.append(item)
151			return result
152
153			def _include_file(self, file_path, content):
154			"""
155			Method which returns True if the following file content should be included in the response.
156
157			Right now we exclude any file with UTF8 BOM character in it - those are most likely binary
158			files such as icon, etc.
159			"""
160			if codecs.BOM_UTF8 in content[:1024]:
161			return False
162
163			if "\0" in content[:1024]:
164			# Found null byte, most likely a binary file
165			return False
166
167			return True
168
169
170			class FileController(BaseFileController):
171			"""
172			Controller which allows user to retrieve content of a specific file in a pack.
173			"""
174
175			def get_one(self, ref_or_id, file_path, requester_user, if_none_match=None,
			0 ignored issues – show Bug introduced 2017-03-24 05:41 UTC by Report Bug Copy Issue Report Arguments number differs from overridden 'get_one' method Loading history...
176			if_modified_since=None):
177			"""
178			Outputs the content of a specific file in a pack.
179
180			Handles requests:
181			GET /packs/views/file/<pack_ref_or_id>/<file path>
182			"""
183			pack_db = self._get_by_ref_or_id(ref_or_id=ref_or_id)
184
185			if not pack_db:
186			msg = 'Pack with ref_or_id "%s" does not exist' % (ref_or_id)
187			raise StackStormDBObjectNotFoundError(msg)
188
189			if not file_path:
190			raise ValueError('Missing file path')
191
192			pack_ref = pack_db.ref
193
194			# Note: Until list filtering is in place we don't require RBAC check for icon file
195			permission_type = PermissionType.PACK_VIEW
196			if file_path not in WHITELISTED_FILE_PATHS:
197			rbac_utils.assert_user_has_resource_db_permission(user_db=requester_user,
198			resource_db=pack_db,
199			permission_type=permission_type)
200
201			normalized_file_path = get_pack_file_abs_path(pack_ref=pack_ref, file_path=file_path)
202			if not normalized_file_path or not os.path.isfile(normalized_file_path):
203			# Ignore references to files which don't exist on disk
204			raise StackStormDBObjectNotFoundError('File "%s" not found' % (file_path))
205
206			file_size, file_mtime = self._get_file_stats(file_path=normalized_file_path)
207
208			response = Response()
209
210			if not self._is_file_changed(file_mtime,
211			if_none_match=if_none_match,
212			if_modified_since=if_modified_since):
213			response.status = http_client.NOT_MODIFIED
214			else:
215			if file_size is not None and file_size > MAX_FILE_SIZE:
216			msg = ('File %s exceeds maximum allowed file size (%s bytes)' %
217			(file_path, MAX_FILE_SIZE))
218			raise ValueError(msg)
219
220			content_type = mimetypes.guess_type(normalized_file_path)[0] or \
221			'application/octet-stream'
222
223			response.headers['Content-Type'] = content_type
224			response.body = self._get_file_content(file_path=normalized_file_path)
225
226			response.headers['Last-Modified'] = format_date_time(file_mtime)
227			response.headers['ETag'] = repr(file_mtime)
228
229			return response
230
231			def _is_file_changed(self, file_mtime, if_none_match=None, if_modified_since=None):
232			# For if_none_match check against what would be the ETAG value
233			if if_none_match:
234			return repr(file_mtime) != if_none_match
235
236			# For if_modified_since check against file_mtime
237			if if_modified_since:
238			return if_modified_since != format_date_time(file_mtime)
239
240			# Neither header is provided therefore assume file is changed.
241			return True
242
243
244			class PackViewsController(object):
245			files = FilesController()
246			file = FileController()
247

StackStorm / st2

Push — develop ( a82689...1ba140 )

FileController.get_one() C

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like