Add feature flag to allow users to be created only if invited to join a team

Allows for open user creation, but only if extended an invitation by someone who already has access

endpoints/oauth/login.py

@@ -12,6 +12,7 @@ from app import app, analytics, get_app_url, oauth_login, authentication
 from auth.auth_context import get_authenticated_user
 from auth.decorators import require_session_login
 from data import model
+from data.users.shared import can_create_user
 from endpoints.common import common_login
 from endpoints.web import index, render_page_template_with_routedata
 from endpoints.csrf import csrf_protect, OAUTH_CSRF_TOKEN_NAME, generate_csrf_token
@@ -86,7 +87,7 @@ def _conduct_oauth_login(auth_system, login_service, lid, lusername, lemail, met
     return _oauthresult(user_obj=user_obj, service_name=service_name)
 
   # Otherwise, we need to create a new user account.
-  if not features.USER_CREATION:
+  if not can_create_user(lemail):
     error_message = 'User creation is disabled. Please contact your administrator'
     return _oauthresult(service_name=service_name, error_message=error_message)
 
@@ -130,7 +131,8 @@ def _conduct_oauth_login(auth_system, login_service, lid, lusername, lemail, met
 def _render_ologin_error(service_name, error_message=None, register_redirect=False):
   """ Returns a Flask response indicating an OAuth error. """
 
-  user_creation = bool(features.USER_CREATION and features.DIRECT_LOGIN)
+  user_creation = bool(features.USER_CREATION and features.DIRECT_LOGIN and
+                       not features.INVITE_ONLY_USER_CREATION)
   error_info = {
     'reason': 'ologinerror',
     'service_name': service_name,

endpoints/oauth/test/test_login.py

@@ -1,5 +1,7 @@
 import pytest
 
+from mock import patch
+
 from data import model, database
 from data.users import get_users_handler, DatabaseUsers
 from endpoints.oauth.login import _conduct_oauth_login
@@ -71,6 +73,37 @@ def test_new_account_via_database(login_service):
   federated_login = model.user.lookup_federated_login(new_user, login_service.service_id())
   assert federated_login is not None
 
+@pytest.mark.parametrize('open_creation, invite_only, has_invite, expect_success', [
+  # Open creation -> Success!
+  (True, False, False, True),
+
+  # Open creation + invite only + no invite -> Failure!
+  (True, True, False, False),
+
+  # Open creation + invite only + invite -> Success!
+  (True, True, True, True),
+
+  # Close creation -> Failure!
+  (False, False, False, False),
+])
+def test_flagged_user_creation(open_creation, invite_only, has_invite, expect_success, login_service):
+  login_service_lid = 'someexternaluser'
+  email = 'some@example.com'
+
+  if has_invite:
+    inviter = model.user.get_user('devtable')
+    team = model.team.get_organization_team('buynlarge', 'owners')
+    model.team.add_or_invite_to_team(inviter, team, email=email)
+
+  internal_auth = DatabaseUsers()
+
+  with patch('features.USER_CREATION', open_creation):
+    with patch('features.INVITE_ONLY_USER_CREATION', invite_only):
+      # Conduct login.
+      result = _conduct_oauth_login(internal_auth, login_service, login_service_lid, login_service_lid,
+                                    email)
+      assert (result.user_obj is not None) == expect_success
+      assert (result.error_message is None) == expect_success
 
 @pytest.mark.parametrize('binding_field, lid, lusername, lemail, expected_error', [
   # No binding field + newly seen user -> New unlinked user