check network access for Request an Account page

changes/1476.a.feature

@@ -0,0 +1 @@
+check network access for account creation

changes/1476.b.feature

@@ -0,0 +1 @@
+check network access for login page

ckanext/canada/auth.py

@@ -1,5 +1,9 @@
-from ckan.plugins.toolkit import chained_auth_function, config
+from ckan.plugins.toolkit import (chained_auth_function,
+                                  auth_allow_anonymous_access,
+                                  config,
+                                  request)
 from ckan.authz import has_user_permission_for_group_or_org, is_sysadmin
+from ckanext.canada.helpers import goc_auth
 
 
 def _is_reporting_user(context):
@@ -27,6 +31,14 @@ def datastore_upsert(up_func, context, data_dict):
     return up_func(context, data_dict)
 
 
+@chained_auth_function
+@auth_allow_anonymous_access
+def user_create(up_func, context, data_dict=None):
+    # additional check to ensure user can access the Request an Account page
+    # only possible if accessing from GOC network
+    goc_auth()
+    return up_func(context, data_dict)
+
 def view_org_members(context, data_dict):
     user = context.get('user')
     can_view = has_user_permission_for_group_or_org(data_dict.get(u'id'), user, 'manage_group')

ckanext/canada/helpers.py

@@ -29,6 +29,8 @@
     from ckanext.xloader.utils import XLoaderFormats
 except ImportError:
     XLoaderFormats = None
+import os
+import ipaddress
 
 
 ORG_MAY_PUBLISH_OPTION = 'canada.publish_datasets_organization_name'
@@ -774,6 +776,52 @@ def date_field(field, pkg):
     return pkg.get(field, None)
 
 
+def goc_auth():
+    if not registry_network_access():
+        return t.abort(403, _(u'This application is only available to authorized '
+                            u'Government of Canada departments and agencies. '
+                            u'Please contact the support team at '
+                            u'[email protected] to request access.'))
+
+
+def registry_network_access():
+    """
+    Only allow requests from GOC network to access
+    user account registration view
+   """
+    remote_addr = request.headers.get('X-Forwarded-For') or \
+                  request.environ.get('REMOTE_ADDR')
+    try:
+        client_ip = ipaddress.ip_address(unicode(remote_addr))
+    except ValueError:
+        return False
+
+    netlist_path = config.get('ckanext.canada.registry_network_list', '')
+    if not netlist_path:
+        return False
+    if not os.path.isfile(netlist_path):
+        return False
+
+    with open(netlist_path) as allow_list:
+        for line in allow_list:
+            # the netlist_path file is a combination of text, ip addresses and subnets
+            line = unicode(line.strip())
+            try:
+                ip = ipaddress.ip_address(line)
+                if client_ip == ip:
+                    return True
+            except ValueError:
+                pass
+
+            try:
+                ip_network = ipaddress.ip_network(line, False)
+                if client_ip in ip_network:
+                    return True
+            except ValueError:
+                pass
+        return False
+
+
 def split_piped_bilingual_field(field_text, client_lang):
     if field_text is not None and ' | ' in field_text:
         return field_text.split(' | ')[1 if client_lang == 'fr' else 0]

ckanext/canada/plugins.py

@@ -68,6 +68,7 @@ class CanadaSecurityPlugin(CkanSecurityPlugin):
     p.implements(p.IResourceController, inherit=True)
     p.implements(p.IValidators, inherit=True)
     p.implements(p.IConfigurer)
+    p.implements(p.IAuthenticator)
 
     def update_config(self, config):
         # Disable auth settings
@@ -102,6 +103,17 @@ def get_validators(self):
                 'canada_security_upload_presence':
                     validators.canada_security_upload_presence}
 
+    # IAuthenticator
+    def identify(self):
+        if p.toolkit.request.path == p.toolkit.url_for(u'user.login'):
+            return helpers.goc_auth()
+
+    def logout(self):
+        return
+
+    def abort(self, status_code, detail, headers, comment):
+        return (status_code, detail, headers, comment)
+
 
 class CanadaDatasetsPlugin(SchemingDatasetsPlugin):
     """
@@ -392,6 +404,7 @@ def get_helpers(self):
             'canada_check_access',
             'get_user_email',
             'get_loader_status_badge',
+            'registry_network_access',
         ])
 
     # IConfigurable
@@ -456,6 +469,7 @@ def get_auth_functions(self):
             'group_show': auth.group_show,
             'organization_list': auth.organization_list,
             'organization_show': auth.organization_show,
+            'user_create': auth.user_create,
         }
 
     # IXloader

ckanext/canada/templates/internal/error_document_template.html

@@ -11,7 +11,7 @@
     <div class="col-md-12">
       <p class="mrgn-tp-lg"><strong>{{ content }}</strong></p>
       {% if code == 403 %}
-        {% if not c.userobj %}
+        {% if not c.userobj and h.registry_network_access() %}
         <p class="mrgn-tp-lg">{{ _('You are not logged in to the Open Government Registry.
           Note that your login expires after 1 hour of inactivity.
           Click on the log in button below to log in.') }}</p>

ckanext/canada/templates/internal/header.html

@@ -31,10 +31,12 @@
         <ul class="list-inline">
             {% block header_account_notlogged %}
             <li>
+              {% if h.registry_network_access() %}
                 <a href="{{ h.url_for('user.login') }}" class="btn btn-primary">
                     <span class="fa fa-sign-in"></span>
                     <span class="text">{{ _('Log in') }}</span>
                 </a>
+              {% endif %}
             </li>
             {% endblock %}
         </ul>

ckanext/canada/tests/factories.py

@@ -19,7 +19,7 @@
 _mock_file_open = fake_filesystem.FakeFileOpen(_fs)
 
 
-def _mock_open_if_open_fails(*args, **kwargs):
+def _mock_open(*args, **kwargs):
     try:
         return real_open(*args, **kwargs)
     except (OSError, IOError):
@@ -30,7 +30,7 @@ def mock_uploads(func):
     @helpers.change_config('ckan.storage_path', '/doesnt_exist')
     @mock.patch.object(ckan.lib.uploader, 'os', _mock_os)
     @mock.patch.object(builtins, 'open',
-                       side_effect=_mock_open_if_open_fails)
+                       side_effect=_mock_open)
     @mock.patch.object(ckan.lib.uploader, '_storage_path',
                        new='/doesnt_exist')
     @functools.wraps(func)

ckanext/canada/tests/test_webforms.py

@@ -1,8 +1,11 @@
 # -*- coding: UTF-8 -*-
+import os
+import mock
 from ckanext.canada.tests import CanadaTestBase
 import pytest
 from urlparse import urlparse
-from StringIO import StringIO
+from cStringIO import StringIO
+from io import StringIO as StringIOWrapper
 from openpyxl.workbook import Workbook
 from ckan.plugins.toolkit import h
 from ckanapi import (
@@ -15,7 +18,8 @@
 from ckan.tests.factories import Sysadmin
 from ckanext.canada.tests.factories import (
     CanadaOrganization as Organization,
-    CanadaUser as User
+    CanadaUser as User,
+    _mock_open,
 )
 
 from ckanext.recombinant.tables import get_chromo
@@ -28,6 +32,23 @@
 )
 
 
+real_isfile = os.path.isfile
+MOCK_IP_ADDRESS = u'174.116.80.148'
+MOCK_IP_LIST_FILE = u'test_ip_list'
+
+
+def _mock_isfile(filename):
+    if MOCK_IP_LIST_FILE in filename:
+        return True
+    return real_isfile(filename)
+
+
+def _mock_open_ip_list(*args, **kwargs):
+    if args and MOCK_IP_LIST_FILE in args[0]:
+        return StringIOWrapper(MOCK_IP_ADDRESS)
+    return _mock_open(*args, **kwargs)
+
+
 def _get_relative_offset_from_response(response):
     # type: (CKANResponse) -> str
     assert response.headers
@@ -182,10 +203,13 @@ def setup_method(self, method):
         Setup any state specific to the execution of the given class methods.
         """
         super(TestNewUserWebForms, self).setup_method(method)
-        self.extra_environ_tester = {'REMOTE_USER': str(u"")}
+        self.extra_environ_tester = {'REMOTE_USER': str(u""), 'REMOTE_ADDR': MOCK_IP_ADDRESS}
+        self.extra_environ_tester_bad_ip = {'REMOTE_USER': str(u""), 'REMOTE_ADDR': '174.116.80.142'}
         self.org = Organization()
 
 
+    @mock.patch('os.path.isfile', _mock_isfile)
+    @mock.patch('__builtin__.open', _mock_open_ip_list)
     def test_new_user_required_fields(self, app):
         offset = h.url_for('user.register')
         response = app.get(offset, extra_environ=self.extra_environ_tester)
@@ -201,6 +225,8 @@ def test_new_user_required_fields(self, app):
         assert 'Thank you for creating your account for the Open Government registry' in response.body
 
 
+    @mock.patch('os.path.isfile', _mock_isfile)
+    @mock.patch('__builtin__.open', _mock_open_ip_list)
     def test_new_user_missing_fields(self, app):
         offset = h.url_for('user.register')
         response = app.get(offset, extra_environ=self.extra_environ_tester)
@@ -224,6 +250,15 @@ def test_new_user_missing_fields(self, app):
         assert 'Password: Please enter both passwords' in response.body
 
 
+    @mock.patch('os.path.isfile', _mock_isfile)
+    @mock.patch('__builtin__.open', _mock_open_ip_list)
+    def test_new_user_bad_ip_address(self, app):
+        offset = h.url_for('user.register')
+        response = app.get(offset, extra_environ=self.extra_environ_tester_bad_ip)
+
+        assert response.status_code == 403
+
+
     def _filled_new_user_form(self):
         # type: () -> dict
         return {

test-core.ini

@@ -33,6 +33,8 @@ ckan.plugins = validation canada_forms canada_internal canada_public
 # no default views for tests...
 # ckan.views.default_views = []
 
+ckanext.canada.registry_network_list = /test_ip_list
+
 # we have tests for web user registration form
 ckan.auth.create_user_via_web = true