2FA (Webauthn)

requirements/main.in

@@ -51,6 +51,7 @@ stdlib-list
 structlog
 transaction
 typeguard
+webauthn
 whitenoise
 WTForms>=2.0.0
 zope.sqlalchemy

requirements/main.txt

@@ -68,6 +68,9 @@ botocore==1.12.161 \
 cachetools==3.1.1 \
     --hash=sha256:428266a1c0d36dc5aca63a2d7c5942e88c2c898d72139fca0e97fdd2380517ae \
     --hash=sha256:8ea2d3ce97850f31e4a08b0e2b5e6c34997d7216a9d2c98e0f3978630d4da69a
+cbor2==4.1.2 \
+    --hash=sha256:17b615da69964f87e48c5adb34ba63db3068f65b9cd14a7b099503d9f8a0e9ae \
+    --hash=sha256:6391fd3d2a4e976ecf892638a0a2a88d85e6764124bf9f128a945bfefefe77dc
 celery-redbeat==0.13.0 \
     --hash=sha256:8ba060f5fffa96fc6dbb0e37a10506cada03333f34ba502b134cbbdb08891266
 celery[sqs]==4.3.0 \
@@ -434,6 +437,9 @@ pycurl==7.43.0.2 \
 pygments==2.4.2 \
     --hash=sha256:71e430bc85c88a430f000ac1d9b331d2407f681d6f6aec95e8bcfbc3df5b0127 \
     --hash=sha256:881c4c157e45f30af185c1ffe8d549d48ac9127433f2c380c24b84572ad66297
+pyOpenSSL==19.0.0 \
+    --hash=sha256:aeca66338f6de19d1aa46ed634c3b9ae519a64b458f8468aec688e7e3c20f200 \
+    --hash=sha256:c727930ad54b10fc157015014b666f2d8b41f70c0d03e83ab67624fd3dd5d1e6
 pyparsing==2.4.0 \
     --hash=sha256:1873c03321fc118f4e9746baf201ff990ceb915f433f23b395f5580d1840cb2a \
     --hash=sha256:9b6323ef4ab914af344ba97510e966d64ba91055d6b9afa6b30799340e89cc03
@@ -532,6 +538,8 @@ venusian==1.2.0 \
 vine==1.3.0 \
     --hash=sha256:133ee6d7a9016f177ddeaf191c1f58421a1dcc6ee9a42c58b34bed40e1d2cd87 \
     --hash=sha256:ea4947cc56d1fd6f2095c8d543ee25dad966f78692528e68b4fada11ba3f98af
+webauthn==0.4.5 \
+    --hash=sha256:db0bfc8b74896e209bfb290815c0a563f785ec5be321e1db17e9ac627ec3562f
 webencodings==0.5.1 \
     --hash=sha256:a0af1213f3c2226497a97e2b3aa01a7e4bee4f403f95be16fc9acd2947514a78 \
     --hash=sha256:b36a1c245f2d304965eb4e0a82848379241dc04b865afcc4aab16748587e1923

tests/unit/accounts/test_forms.py

@@ -10,13 +10,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
+
 import pretend
 import pytest
 import wtforms
 
 from warehouse.accounts import forms
 from warehouse.accounts.interfaces import TooManyFailedLogins
 from warehouse.accounts.models import DisableReason
+from warehouse.utils.webauthn import AuthenticationRejectedException
 
 
 class TestLoginForm:
@@ -555,40 +558,106 @@ def test_password_breached(self):
         )
 
 
-class TestTwoFactorForm:
+class TestTOTPAuthenticationForm:
     def test_creation(self):
         user_id = pretend.stub()
         user_service = pretend.stub()
-        form = forms.TwoFactorForm(user_id=user_id, user_service=user_service)
+        form = forms.TOTPAuthenticationForm(user_id=user_id, user_service=user_service)
 
         assert form.user_service is user_service
 
     def test_totp_secret_exists(self):
-        form = forms.TwoFactorForm(
+        form = forms.TOTPAuthenticationForm(
             data={"totp_value": ""}, user_id=pretend.stub(), user_service=pretend.stub()
         )
         assert not form.validate()
         assert form.totp_value.errors.pop() == "This field is required."
 
-        form = forms.TwoFactorForm(
+        form = forms.TOTPAuthenticationForm(
             data={"totp_value": "not_a_real_value"},
             user_id=pretend.stub(),
             user_service=pretend.stub(check_totp_value=lambda *a: True),
         )
         assert not form.validate()
         assert form.totp_value.errors.pop() == "TOTP code must be 6 digits."
 
-        form = forms.TwoFactorForm(
+        form = forms.TOTPAuthenticationForm(
             data={"totp_value": "123456"},
             user_id=pretend.stub(),
             user_service=pretend.stub(check_totp_value=lambda *a: False),
         )
         assert not form.validate()
         assert form.totp_value.errors.pop() == "Invalid TOTP code."
 
-        form = forms.TwoFactorForm(
+        form = forms.TOTPAuthenticationForm(
             data={"totp_value": "123456"},
             user_id=pretend.stub(),
             user_service=pretend.stub(check_totp_value=lambda *a: True),
         )
         assert form.validate()
+
+
+class TestWebAuthnAuthenticationForm:
+    def test_creation(self):
+        user_id = pretend.stub()
+        user_service = pretend.stub()
+        challenge = pretend.stub()
+        origin = pretend.stub()
+        icon_url = pretend.stub()
+        rp_id = pretend.stub()
+
+        form = forms.WebAuthnAuthenticationForm(
+            user_id=user_id,
+            user_service=user_service,
+            challenge=challenge,
+            origin=origin,
+            icon_url=icon_url,
+            rp_id=rp_id,
+        )
+
+        assert form.challenge is challenge
+
+    def test_credential_bad_payload(self):
+        form = forms.WebAuthnAuthenticationForm(
+            credential="not valid json",
+            user_id=pretend.stub(),
+            user_service=pretend.stub(),
+            challenge=pretend.stub(),
+            origin=pretend.stub(),
+            icon_url=pretend.stub(),
+            rp_id=pretend.stub(),
+        )
+        assert not form.validate()
+        assert form.credential.errors.pop() == "Invalid WebAuthn assertion: Bad payload"
+
+    def test_credential_invalid(self):
+        form = forms.WebAuthnAuthenticationForm(
+            credential=json.dumps({}),
+            user_id=pretend.stub(),
+            user_service=pretend.stub(
+                verify_webauthn_assertion=pretend.raiser(
+                    AuthenticationRejectedException("foo")
+                )
+            ),
+            challenge=pretend.stub(),
+            origin=pretend.stub(),
+            icon_url=pretend.stub(),
+            rp_id=pretend.stub(),
+        )
+        assert not form.validate()
+        assert form.credential.errors.pop() == "foo"
+
+    def test_credential_valid(self):
+        form = forms.WebAuthnAuthenticationForm(
+            credential=json.dumps({}),
+            user_id=pretend.stub(),
+            user_service=pretend.stub(
+                verify_webauthn_assertion=pretend.call_recorder(lambda *a, **kw: 123456)
+            ),
+            challenge=pretend.stub(),
+            origin=pretend.stub(),
+            icon_url=pretend.stub(),
+            rp_id=pretend.stub(),
+        )
+        assert form.validate()
+        assert form.sign_count == 123456

tests/unit/accounts/test_services.py

@@ -22,6 +22,7 @@
 from zope.interface.verify import verifyClass
 
 import warehouse.utils.otp as otp
+import warehouse.utils.webauthn as webauthn
 
 from warehouse.accounts import services
 from warehouse.accounts.interfaces import (
@@ -351,6 +352,20 @@ def test_has_two_factor(self, user_service):
         user_service.update_user(user.id, totp_secret=b"foobar")
         assert user_service.has_two_factor(user.id)
 
+    def test_has_totp(self, user_service):
+        user = UserFactory.create()
+        assert not user_service.has_totp(user.id)
+        user_service.update_user(user.id, totp_secret=b"foobar")
+        assert user_service.has_totp(user.id)
+
+    def test_has_webauthn(self, user_service):
+        user = UserFactory.create()
+        assert not user_service.has_webauthn(user.id)
+        user_service.add_webauthn(
+            user.id, credential_id="foo", public_key="bar", sign_count=1
+        )
+        assert user_service.has_webauthn(user.id)
+
     @pytest.mark.parametrize("valid", [True, False])
     def test_check_totp_value(self, user_service, monkeypatch, valid):
         verify_totp = pretend.call_recorder(lambda *a: valid)
@@ -406,6 +421,133 @@ def test_check_totp_value_user_rate_limited(self, user_service, metrics):
             ),
         ]
 
+    @pytest.mark.parametrize(
+        ("challenge", "rp_name", "rp_id", "icon_url"),
+        (
+            ["fake_challenge", "fake_rp_name", "fake_rp_id", "fake_icon_url"],
+            [None, None, None, None],
+        ),
+    )
+    def test_get_webauthn_credential_options(
+        self, user_service, challenge, rp_name, rp_id, icon_url
+    ):
+        user = UserFactory.create()
+        options = user_service.get_webauthn_credential_options(
+            user.id,
+            challenge=challenge,
+            rp_name=rp_name,
+            rp_id=rp_id,
+            icon_url=icon_url,
+        )
+
+        assert options["user"]["id"] == str(user.id)
+        assert options["user"]["name"] == user.username
+        assert options["user"]["displayName"] == user.name
+        assert options["challenge"] == challenge
+        assert options["rp"]["name"] == rp_name
+        assert options["rp"]["id"] == rp_id
+
+        if icon_url:
+            assert options["user"]["icon"] == icon_url
+        else:
+            assert "icon" not in options["user"]
+
+    def test_get_webauthn_assertion_options(self, user_service):
+        user = UserFactory.create()
+        user_service.add_webauthn(
+            user.id, credential_id="foo", public_key="bar", sign_count=1
+        )
+
+        options = user_service.get_webauthn_assertion_options(
+            user.id,
+            challenge="fake_challenge",
+            icon_url="fake_icon_url",
+            rp_id="fake_rp_id",
+        )
+
+        assert options["challenge"] == "fake_challenge"
+        assert options["rpId"] == "fake_rp_id"
+        assert options["allowCredentials"][0]["id"] == user.webauthn.credential_id
+
+    def test_verify_webauthn_credential(self, user_service, monkeypatch):
+        user = UserFactory.create()
+        user_service.add_webauthn(
+            user.id, credential_id="foo", public_key="bar", sign_count=1
+        )
+
+        fake_validated_credential = pretend.stub(credential_id=b"bar")
+        verify_registration_response = pretend.call_recorder(
+            lambda *a, **kw: fake_validated_credential
+        )
+        monkeypatch.setattr(
+            webauthn, "verify_registration_response", verify_registration_response
+        )
+
+        validated_credential = user_service.verify_webauthn_credential(
+            pretend.stub(),
+            challenge=pretend.stub(),
+            rp_id=pretend.stub(),
+            origin=pretend.stub(),
+        )
+
+        assert validated_credential is fake_validated_credential
+
+    def test_verify_webauthn_credential_already_in_use(self, user_service, monkeypatch):
+        user = UserFactory.create()
+        user_service.add_webauthn(
+            user.id, credential_id="foo", public_key="bar", sign_count=1
+        )
+
+        fake_validated_credential = pretend.stub(credential_id=b"foo")
+        verify_registration_response = pretend.call_recorder(
+            lambda *a, **kw: fake_validated_credential
+        )
+        monkeypatch.setattr(
+            webauthn, "verify_registration_response", verify_registration_response
+        )
+
+        with pytest.raises(webauthn.RegistrationRejectedException):
+            user_service.verify_webauthn_credential(
+                pretend.stub(),
+                challenge=pretend.stub(),
+                rp_id=pretend.stub(),
+                origin=pretend.stub(),
+            )
+
+    def test_verify_webauthn_assertion(self, user_service, monkeypatch):
+        user = UserFactory.create()
+        user_service.add_webauthn(
+            user.id, credential_id="foo", public_key="bar", sign_count=1
+        )
+
+        verify_assertion_response = pretend.call_recorder(lambda *a, **kw: 2)
+        monkeypatch.setattr(
+            webauthn, "verify_assertion_response", verify_assertion_response
+        )
+
+        updated_sign_count = user_service.verify_webauthn_assertion(
+            user.id,
+            pretend.stub(),
+            challenge=pretend.stub(),
+            origin=pretend.stub(),
+            icon_url=pretend.stub(),
+            rp_id=pretend.stub(),
+        )
+        assert updated_sign_count == 2
+
+    def test_add_webauthn_already_added(self, user_service):
+        user = UserFactory.create()
+        user_service.add_webauthn(
+            user.id, credential_id="foo", public_key="bar", sign_count=1
+        )
+
+        assert (
+            user_service.add_webauthn(
+                user.id, credential_id="baz", public_key="quux", sign_count=1337
+            )
+            is None
+        )
+
 
 class TestTokenService:
     def test_verify_service(self):