diff --git a/google/oauth2/webauthn_handler.py b/google/oauth2/webauthn_handler.py
new file mode 100644
index 000000000..eb4fa2662
--- /dev/null
+++ b/google/oauth2/webauthn_handler.py
@@ -0,0 +1,82 @@
+import abc
+import os
+import struct
+import subprocess
+
+from google.auth import exceptions
+from google.oauth2.webauthn_types import GetRequest, GetResponse
+
+
+class WebAuthnHandler(abc.ABC):
+    @abc.abstractmethod
+    def is_available(self) -> bool:
+        """Check whether this WebAuthn handler is available"""
+        raise NotImplementedError("is_available method must be implemented")
+
+    @abc.abstractmethod
+    def get(self, get_request: GetRequest) -> GetResponse:
+        """WebAuthn get (assertion)"""
+        raise NotImplementedError("get method must be implemented")
+
+
+class PluginHandler(WebAuthnHandler):
+    """Offloads WebAuthn get reqeust to a pluggable command-line tool.
+
+    Offloads WebAuthn get to a plugin which takes the form of a
+    command-line tool. The command-line tool is configurable via the
+    PluginHandler._ENV_VAR environment variable.
+
+    The WebAuthn plugin should implement the following interface:
+
+    Communication occurs over stdin/stdout, and messages are both sent and
+    received in the form:
+
+    [4 bytes - payload size (little-endian)][variable bytes - json payload]
+    """
+
+    _ENV_VAR = "GOOGLE_AUTH_WEBAUTHN_PLUGIN"
+
+    def is_available(self) -> bool:
+        try:
+            self._find_plugin()
+        except:
+            return False
+        else:
+            return True
+
+    def get(self, get_request: GetRequest) -> GetResponse:
+        request_json = get_request.to_json()
+        cmd = self._find_plugin()
+        response_json = self._call_plugin(cmd, request_json)
+        return GetResponse.from_json(response_json)
+
+    def _call_plugin(self, cmd: str, input_json: str) -> str:
+        # Calculate length of input
+        input_length = len(input_json)
+        length_bytes_le = struct.pack("<I", input_length)
+        request = length_bytes_le + input_json.encode()
+
+        # Call plugin
+        process_result = subprocess.run(
+            [cmd], input=request, capture_output=True, check=True
+        )
+
+        # Check length of response
+        response_len_le = process_result.stdout[:4]
+        response_len = struct.unpack("<I", response_len_le)[0]
+        response = process_result.stdout[4:]
+        if response_len != len(response):
+            raise exceptions.MalformedError(
+                "Plugin response length {} does not match data {}".format(
+                    response_len, len(response)
+                )
+            )
+        return response.decode()
+
+    def _find_plugin(self) -> str:
+        plugin_cmd = os.environ.get(PluginHandler._ENV_VAR)
+        if plugin_cmd is None:
+            raise exceptions.InvalidResource(
+                "{} env var is not set".format(PluginHandler._ENV_VAR)
+            )
+        return plugin_cmd
diff --git a/google/oauth2/webauthn_types.py b/google/oauth2/webauthn_types.py
new file mode 100644
index 000000000..92c56a559
--- /dev/null
+++ b/google/oauth2/webauthn_types.py
@@ -0,0 +1,160 @@
+from dataclasses import dataclass
+import json
+from typing import List, Dict, Optional
+
+from google.auth import exceptions
+
+
+@dataclass(frozen=True)
+class PublicKeyCredentialDescriptor:
+    """Descriptor for a security key based credential.
+
+    https://www.w3.org/TR/webauthn-3/#dictionary-credential-descriptor
+
+    Args:
+        id: <url-safe base64-encoded> credential id (key handle).
+        transports: <'usb'|'nfc'|'ble'|'internal'> List of supported transports.
+    """
+
+    id: str
+    transports: Optional[List[str]] = None
+
+    def to_dict(self):
+        cred = {"type": "public-key", "id": self.id}
+        if self.transports:
+            cred["transports"] = self.transports
+        return cred
+
+
+@dataclass
+class AuthenticationExtensionsClientInputs:
+    """Client extensions inputs for WebAuthn extensions.
+
+    Args:
+        appid: app id that can be asserted with in addition to rpid.
+            https://www.w3.org/TR/webauthn-3/#sctn-appid-extension
+    """
+
+    appid: Optional[str] = None
+
+    def to_dict(self):
+        extensions = {}
+        if self.appid:
+            extensions["appid"] = self.appid
+        return extensions
+
+
+@dataclass
+class GetRequest:
+    """WebAuthn get request
+
+    Args:
+        origin: Origin where the WebAuthn get assertion takes place.
+        rpid: Relying Party ID.
+        challenge: <url-safe base64-encoded> raw challenge.
+        timeout_ms: Timeout number in millisecond.
+        allow_credentials: List of allowed credentials.
+        user_verification: <'required'|'preferred'|'discouraged'> User verification requirement.
+        extensions: WebAuthn authentication extensions inputs.
+    """
+
+    origin: str
+    rpid: str
+    challenge: str
+    timeout_ms: Optional[int] = None
+    allow_credentials: Optional[List[PublicKeyCredentialDescriptor]] = None
+    user_verification: Optional[str] = None
+    extensions: Optional[AuthenticationExtensionsClientInputs] = None
+
+    def to_json(self) -> str:
+        req_options = {"rpid": self.rpid, "challenge": self.challenge}
+        if self.timeout_ms:
+            req_options["timeout"] = self.timeout_ms
+        if self.allow_credentials:
+            req_options["allowCredentials"] = [
+                c.to_dict() for c in self.allow_credentials
+            ]
+        if self.user_verification:
+            req_options["userVerification"] = self.user_verification
+        if self.extensions:
+            req_options["extensions"] = self.extensions.to_dict()
+        return json.dumps(
+            {
+                "type": "get",
+                "origin": self.origin,
+                "requestData": req_options,
+            }
+        )
+
+
+@dataclass(frozen=True)
+class AuthenticatorAssertionResponse:
+    """Authenticator response to a WebAuthn get (assertion) request.
+
+    https://www.w3.org/TR/webauthn-3/#authenticatorassertionresponse
+
+    Args:
+        client_data_json: <url-safe base64-encoded> client data JSON.
+        authenticator_data: <url-safe base64-encoded> authenticator data.
+        signature: <url-safe base64-encoded> signature.
+        user_handle: <url-safe base64-encoded> user handle.
+    """
+
+    client_data_json: str
+    authenticator_data: str
+    signature: str
+    user_handle: Optional[str]
+
+
+@dataclass(frozen=True)
+class GetResponse:
+    """WebAuthn get (assertion) response.
+
+    Args:
+        id: <url-safe base64-encoded> credential id (key handle).
+        response: The authenticator assertion response.
+        authenticator_attachment: <'cross-platform'|'platform'> The attachment status of the authenticator.
+        client_extension_results: WebAuthn authentication extensions output results in a dictionary.
+    """
+
+    id: str
+    response: AuthenticatorAssertionResponse
+    authenticator_attachment: Optional[str]
+    client_extension_results: Optional[Dict]
+
+    @staticmethod
+    def from_json(json_str: str):
+        """Verify and construct GetResponse from a JSON string."""
+        try:
+            resp_json = json.loads(json_str)
+        except ValueError:
+            raise exceptions.MalformedError("Invalid Get JSON response")
+        if resp_json.get("type") != "getResponse":
+            raise exceptions.MalformedError(
+                "Invalid Get response type: {}".format(resp_json.get("type"))
+            )
+        pk_cred = resp_json.get("responseData")
+        if pk_cred is None:
+            if resp_json.get("error"):
+                raise exceptions.ReauthFailError(
+                    "WebAuthn.get failure: {}".format(resp_json["error"])
+                )
+            else:
+                raise exceptions.MalformedError("Get response is empty")
+        if pk_cred.get("type") != "public-key":
+            raise exceptions.MalformedError(
+                "Invalid credential type: {}".format(pk_cred.get("type"))
+            )
+        assertion_json = pk_cred["response"]
+        assertion_resp = AuthenticatorAssertionResponse(
+            client_data_json=assertion_json["clientDataJSON"],
+            authenticator_data=assertion_json["authenticatorData"],
+            signature=assertion_json["signature"],
+            user_handle=assertion_json.get("userHandle"),
+        )
+        return GetResponse(
+            id=pk_cred["id"],
+            response=assertion_resp,
+            authenticator_attachment=pk_cred.get("authenticatorAttachment"),
+            client_extension_results=pk_cred.get("clientExtensionResults"),
+        )
diff --git a/tests/oauth2/test_webauthn_handler.py b/tests/oauth2/test_webauthn_handler.py
new file mode 100644
index 000000000..55d72e7e6
--- /dev/null
+++ b/tests/oauth2/test_webauthn_handler.py
@@ -0,0 +1,156 @@
+import json
+import mock
+import pytest
+import struct
+
+from google.auth import exceptions
+from google.oauth2 import webauthn_handler
+from google.oauth2 import webauthn_types
+
+
+@pytest.fixture
+def os_get_stub():
+    with mock.patch.object(
+        webauthn_handler.os.environ,
+        "get",
+        return_value="gcloud_webauthn_plugin",
+        name="fake os.environ.get",
+    ) as mock_os_environ_get:
+        yield mock_os_environ_get
+
+
+@pytest.fixture
+def subprocess_run_stub():
+    with mock.patch.object(
+        webauthn_handler.subprocess,
+        "run",
+        name="fake subprocess.run",
+    ) as mock_subprocess_run:
+        yield mock_subprocess_run
+
+
+def test_PluginHandler_is_available(os_get_stub):
+    test_handler = webauthn_handler.PluginHandler()
+
+    assert test_handler.is_available() == True
+
+    os_get_stub.return_value = None
+    assert test_handler.is_available() == False
+
+
+GET_ASSERTION_REQUEST = webauthn_types.GetRequest(
+    origin="fake_origin",
+    rpid="fake_rpid",
+    challenge="fake_challenge",
+    allow_credentials=[
+        webauthn_types.PublicKeyCredentialDescriptor(id="fake_id_1"),
+    ],
+)
+
+
+def test_malformated_get_assertion_response(os_get_stub, subprocess_run_stub):
+    response_len = struct.pack("<I", 5)
+    response = "1234567890"
+    mock_response = mock.Mock()
+    mock_response.stdout = response_len + response.encode()
+    subprocess_run_stub.return_value = mock_response
+
+    test_handler = webauthn_handler.PluginHandler()
+    with pytest.raises(exceptions.MalformedError) as excinfo:
+        test_handler.get(GET_ASSERTION_REQUEST)
+    assert "Plugin response length" in str(excinfo.value)
+
+
+def test_failure_get_assertion(os_get_stub, subprocess_run_stub):
+    failure_response = {
+        "type": "getResponse",
+        "error": "fake_plugin_get_assertion_failure",
+    }
+    response_json = json.dumps(failure_response).encode()
+    response_len = struct.pack("<I", len(response_json))
+
+    # process returns get response in json
+    mock_response = mock.Mock()
+    mock_response.stdout = response_len + response_json
+    subprocess_run_stub.return_value = mock_response
+
+    test_handler = webauthn_handler.PluginHandler()
+    with pytest.raises(exceptions.ReauthFailError) as excinfo:
+        test_handler.get(GET_ASSERTION_REQUEST)
+    assert failure_response["error"] in str(excinfo.value)
+
+
+def test_success_get_assertion(os_get_stub, subprocess_run_stub):
+    success_response = {
+        "type": "public-key",
+        "id": "fake-id",
+        "authenticatorAttachment": "cross-platform",
+        "clientExtensionResults": {"appid": True},
+        "response": {
+            "clientDataJSON": "fake_client_data_json_base64",
+            "authenticatorData": "fake_authenticator_data_base64",
+            "signature": "fake_signature_base64",
+            "userHandle": "fake_user_handle_base64",
+        },
+    }
+    valid_plugin_response = {"type": "getResponse", "responseData": success_response}
+    valid_plugin_response_json = json.dumps(valid_plugin_response).encode()
+    valid_plugin_response_len = struct.pack("<I", len(valid_plugin_response_json))
+
+    # process returns get response in json
+    mock_response = mock.Mock()
+    mock_response.stdout = valid_plugin_response_len + valid_plugin_response_json
+    subprocess_run_stub.return_value = mock_response
+
+    # Call get()
+    test_handler = webauthn_handler.PluginHandler()
+    got_response = test_handler.get(GET_ASSERTION_REQUEST)
+
+    # Validate expected plugin request
+    os_get_stub.assert_called_once()
+    subprocess_run_stub.assert_called_once()
+
+    stdin_input = subprocess_run_stub.call_args.kwargs["input"]
+    input_json_len_le = stdin_input[:4]
+    input_json_len = struct.unpack("<I", input_json_len_le)[0]
+    input_json = stdin_input[4:]
+    assert len(input_json) == input_json_len
+
+    input_dict = json.loads(input_json.decode("utf8"))
+    assert input_dict == {
+        "type": "get",
+        "origin": "fake_origin",
+        "requestData": {
+            "rpid": "fake_rpid",
+            "challenge": "fake_challenge",
+            "allowCredentials": [
+                {
+                    "type": "public-key",
+                    "id": "fake_id_1",
+                }
+            ],
+        },
+    }
+
+    # Validate get assertion response
+    assert got_response.id == success_response["id"]
+    assert (
+        got_response.authenticator_attachment
+        == success_response["authenticatorAttachment"]
+    )
+    assert (
+        got_response.client_extension_results
+        == success_response["clientExtensionResults"]
+    )
+    assert (
+        got_response.response.client_data_json
+        == success_response["response"]["clientDataJSON"]
+    )
+    assert (
+        got_response.response.authenticator_data
+        == success_response["response"]["authenticatorData"]
+    )
+    assert got_response.response.signature == success_response["response"]["signature"]
+    assert (
+        got_response.response.user_handle == success_response["response"]["userHandle"]
+    )
diff --git a/tests/oauth2/test_webauthn_types.py b/tests/oauth2/test_webauthn_types.py
new file mode 100644
index 000000000..8d7bbe91a
--- /dev/null
+++ b/tests/oauth2/test_webauthn_types.py
@@ -0,0 +1,258 @@
+import json
+import pytest
+
+from google.oauth2 import webauthn_types
+
+
+@pytest.mark.parametrize(
+    "test_pub_key_cred,expected_dict",
+    [
+        (
+            webauthn_types.PublicKeyCredentialDescriptor(
+                id="fake_cred_id_base64", transports=None
+            ),
+            {
+                "type": "public-key",
+                "id": "fake_cred_id_base64",
+            },
+        ),
+        (
+            webauthn_types.PublicKeyCredentialDescriptor(
+                id="fake_cred_id_base64", transports=[]
+            ),
+            {
+                "type": "public-key",
+                "id": "fake_cred_id_base64",
+            },
+        ),
+        (
+            webauthn_types.PublicKeyCredentialDescriptor(
+                id="fake_cred_id_base64", transports=["usb"]
+            ),
+            {
+                "type": "public-key",
+                "id": "fake_cred_id_base64",
+                "transports": ["usb"],
+            },
+        ),
+        (
+            webauthn_types.PublicKeyCredentialDescriptor(
+                id="fake_cred_id_base64", transports=["usb", "internal"]
+            ),
+            {
+                "type": "public-key",
+                "id": "fake_cred_id_base64",
+                "transports": ["usb", "internal"],
+            },
+        ),
+    ],
+)
+def test_PublicKeyCredentialDescriptor(test_pub_key_cred, expected_dict):
+    assert test_pub_key_cred.to_dict() == expected_dict
+
+
+@pytest.mark.parametrize(
+    "test_extension_input,expected_dict",
+    [
+        (
+            webauthn_types.AuthenticationExtensionsClientInputs(),
+            {},
+        ),
+        (
+            webauthn_types.AuthenticationExtensionsClientInputs(appid=""),
+            {},
+        ),
+        (
+            webauthn_types.AuthenticationExtensionsClientInputs(appid="fake_appid"),
+            {"appid": "fake_appid"},
+        ),
+    ],
+)
+def test_AuthenticationExtensionsClientInputs(test_extension_input, expected_dict):
+    assert test_extension_input.to_dict() == expected_dict
+
+
+def test_GetRequest():
+    test_get_request = webauthn_types.GetRequest(
+        origin="fake_origin",
+        rpid="fake_rpid",
+        challenge="fake_challenge",
+        timeout_ms=123,
+        allow_credentials=[
+            webauthn_types.PublicKeyCredentialDescriptor(id="fake_id_1"),
+            webauthn_types.PublicKeyCredentialDescriptor(id="fake_id_2"),
+        ],
+        user_verification="preferred",
+        extensions=webauthn_types.AuthenticationExtensionsClientInputs(
+            appid="fake_appid"
+        ),
+    )
+    assert json.loads(test_get_request.to_json()) == {
+        "type": "get",
+        "origin": "fake_origin",
+        "requestData": {
+            "rpid": "fake_rpid",
+            "timeout": 123,
+            "challenge": "fake_challenge",
+            "userVerification": "preferred",
+            "allowCredentials": [
+                {
+                    "type": "public-key",
+                    "id": "fake_id_1",
+                },
+                {
+                    "type": "public-key",
+                    "id": "fake_id_2",
+                },
+            ],
+            "extensions": {"appid": "fake_appid"},
+        },
+    }
+
+
+@pytest.mark.parametrize(
+    "has_user_handle,has_authenticator_attachment,has_client_extension_results",
+    [
+        (False, False, False),
+        (False, False, True),
+        (False, True, False),
+        (False, True, True),
+        (True, False, False),
+        (True, False, True),
+        (True, True, False),
+        (True, True, True),
+    ],
+)
+def test_GetResponse(
+    has_user_handle, has_authenticator_attachment, has_client_extension_results
+):
+    input_response_data = {
+        "type": "public-key",
+        "id": "fake-id",
+        "authenticatorAttachment": "cross-platform",
+        "clientExtensionResults": {"appid": True},
+        "response": {
+            "clientDataJSON": "fake_client_data_json_base64",
+            "authenticatorData": "fake_authenticator_data_base64",
+            "signature": "fake_signature_base64",
+            "userHandle": "fake_user_handle_base64",
+        },
+    }
+    if not has_authenticator_attachment:
+        input_response_data.pop("authenticatorAttachment")
+    if not has_client_extension_results:
+        input_response_data.pop("clientExtensionResults")
+    if not has_user_handle:
+        input_response_data["response"].pop("userHandle")
+
+    response = webauthn_types.GetResponse.from_json(
+        json.dumps({"type": "getResponse", "responseData": input_response_data})
+    )
+
+    assert response.id == input_response_data["id"]
+    assert response.authenticator_attachment == (
+        input_response_data["authenticatorAttachment"]
+        if has_authenticator_attachment
+        else None
+    )
+    assert response.client_extension_results == (
+        input_response_data["clientExtensionResults"]
+        if has_client_extension_results
+        else None
+    )
+    assert (
+        response.response.client_data_json
+        == input_response_data["response"]["clientDataJSON"]
+    )
+    assert (
+        response.response.authenticator_data
+        == input_response_data["response"]["authenticatorData"]
+    )
+    assert response.response.signature == input_response_data["response"]["signature"]
+    assert response.response.user_handle == (
+        input_response_data["response"]["userHandle"] if has_user_handle else None
+    )
+
+
+@pytest.mark.parametrize(
+    "input_dict,expected_error",
+    [
+        (
+            {"xyz_type": "wrong_type"},
+            "Invalid Get response type",
+        ),
+        (
+            {"type": "wrong_type"},
+            "Invalid Get response type",
+        ),
+        (
+            {"type": "getResponse"},
+            "Get response is empty",
+        ),
+        (
+            {"type": "getResponse", "error": "fake_get_response_error"},
+            "WebAuthn.get failure: fake_get_response_error",
+        ),
+        (
+            {"type": "getResponse", "responseData": {"xyz_type": "wrong_type"}},
+            "Invalid credential type",
+        ),
+        (
+            {"type": "getResponse", "responseData": {"type": "wrong_type"}},
+            "Invalid credential type",
+        ),
+        (
+            {
+                "type": "getResponse",
+                "responseData": {"type": "public-key", "response": {}},
+            },
+            "KeyError",
+        ),
+        (
+            {
+                "type": "getResponse",
+                "responseData": {
+                    "type": "public-key",
+                    "response": {
+                        "clientDataJSON": "fake_client_data_json_base64",
+                    },
+                },
+            },
+            "KeyError",
+        ),
+        (
+            {
+                "type": "getResponse",
+                "responseData": {
+                    "type": "public-key",
+                    "response": {
+                        "clientDataJSON": "fake_client_data_json_base64",
+                        "authenticatorData": "fake_authenticator_data_base64",
+                    },
+                },
+            },
+            "KeyError",
+        ),
+        (
+            {
+                "type": "getResponse",
+                "responseData": {
+                    "type": "public-key",
+                    "response": {
+                        "clientDataJSON": "fake_client_data_json_base64",
+                        "authenticatorData": "fake_authenticator_data_base64",
+                        "signature": "fake_signature_base64",
+                    },
+                },
+            },
+            "KeyError",
+        ),
+    ],
+)
+def test_GetResponse_error(input_dict, expected_error):
+    with pytest.raises(Exception) as excinfo:
+        webauthn_types.GetResponse.from_json(json.dumps(input_dict))
+    if expected_error == "KeyError":
+        assert excinfo.type is KeyError
+    else:
+        assert expected_error in str(excinfo.value)