Bearer token auth

docs/guide/scripting.md

@@ -94,6 +94,7 @@ def on_request(request: RequestModel, posting: Posting) -> None:
     # Set auth on the request.
     request.auth = Auth.basic_auth("username", "password")
     # request.auth = Auth.digest_auth("username", "password")
+    # request.auth = Auth.bearer_token_auth("token")
 
     # This will be captured and written to the log.
     print("Request is being sent!")

docs/roadmap.md

@@ -22,7 +22,7 @@ If you have any feedback or suggestions, please open a new discussion on GitHub.
 - Changing the environment at runtime - probably via command palette - push a new command palette screen where you can search for and select one of the previously used environments.
 - Variable completion autocompletion TextAreas.
 - Variable resolution highlighting in TextAreas.
-- Bearer token auth (can be done now by adding header).
+- Bearer token auth (can be done now by adding header). ✅
 - API key auth (can be done now by adding header).
 - OAuth2 (need to scope out what's involved here).
 - Add "quit" to command palette and footer ✅

pyproject.toml

@@ -10,6 +10,7 @@ dependencies = [
     "xdg-base-dirs>=6.0.1,<7.0.0",
     "click-default-group>=1.2.4,<2.0.0",
     "httpx[brotli]>=0.27.2,<1.0.0",
+    "openapi-pydantic>=0.5.0",
     "pyperclip>=1.9.0,<2.0.0",
     "pydantic>=2.9.2,<3.0.0",
     "pyyaml>=6.0.2,<7.0.0",

src/posting/auth.py

@@ -0,0 +1,12 @@
+from typing import Generator
+
+import httpx
+
+
+class HttpxBearerTokenAuth(httpx.Auth):
+    def __init__(self, token: str):
+        self.token = token
+
+    def auth_flow(self, request: httpx.Request) -> Generator[httpx.Request, httpx.Response, None]:
+        request.headers["Authorization"] = f"Bearer {self.token}"
+        yield request

src/posting/collection.py

@@ -8,6 +8,7 @@
 import rich
 import yaml
 import os
+from posting.auth import HttpxBearerTokenAuth
 from posting.tuple_to_multidict import tuples_to_dict
 from posting.variables import SubstitutionError
 
@@ -34,9 +35,10 @@ def str_presenter(dumper, data):
 
 
 class Auth(BaseModel):
-    type: Literal["basic", "digest"] | None = Field(default=None)
+    type: Literal["basic", "digest", "bearer_token"] | None = Field(default=None)
     basic: BasicAuth | None = Field(default=None)
     digest: DigestAuth | None = Field(default=None)
+    bearer_token: BearerTokenAuth | None = Field(default=None)
 
     def to_httpx_auth(self) -> httpx.Auth | None:
         if self.type == "basic":
@@ -45,6 +47,9 @@ def to_httpx_auth(self) -> httpx.Auth | None:
         elif self.type == "digest":
             assert self.digest is not None
             return httpx.DigestAuth(self.digest.username, self.digest.password)
+        elif self.type == "bearer_token":
+            assert self.bearer_token is not None
+            return HttpxBearerTokenAuth(self.bearer_token.token)
         return None
 
     @classmethod
@@ -57,6 +62,10 @@ def digest_auth(cls, username: str, password: str) -> Auth:
             type="digest", digest=DigestAuth(username=username, password=password)
         )
 
+    @classmethod
+    def bearer_token_auth(cls, token: str) -> Auth:
+        return cls(type="bearer_token", bearer_token=BearerTokenAuth(token=token))
+
 
 class BasicAuth(BaseModel):
     username: str = Field(default="")
@@ -68,6 +77,11 @@ class DigestAuth(BaseModel):
     password: str = Field(default="")
 
 
+
+class BearerTokenAuth(BaseModel):
+    token: str = Field(default="")
+
+
 class Header(BaseModel):
     name: str
     value: str
@@ -239,6 +253,9 @@ def apply_template(self, variables: dict[str, Any]) -> None:
                     self.auth.digest.username = template.substitute(variables)
                     template = Template(self.auth.digest.password)
                     self.auth.digest.password = template.substitute(variables)
+                if self.auth.bearer_token is not None:
+                    template = Template(self.auth.bearer_token.token)
+                    self.auth.bearer_token.token = template.substitute(variables)
         except (KeyError, ValueError) as e:
             raise SubstitutionError(f"Variable not defined: {e}")
 

src/posting/importing/curl.py

@@ -218,6 +218,15 @@ def _extract_auth_from_headers(self) -> tuple[Auth | None, list[tuple[str, str]]
                     except Exception:
                         # If we can't parse it, keep it as a header
                         remaining_headers.append((name, value))
+
+                elif auth_type_lower == "bearer":
+                    # Bearer token auth
+                    try:
+                        auth = Auth.bearer_token_auth(auth_value)
+                    except Exception:
+                        # If we can't parse it, keep it as a header
+                        remaining_headers.append((name, value))
+
                 else:
                     # Unknown auth type, keep as header
                     remaining_headers.append((name, value))

src/posting/importing/open_api.py

@@ -4,12 +4,16 @@
 from urllib.parse import urlparse
 
 import yaml
+from openapi_pydantic import OpenAPI, Reference, SecurityScheme
 from pathlib import Path
 
 
 from posting.collection import (
     VALID_HTTP_METHODS,
     APIInfo,
+    Auth,
+    BasicAuth,
+    BearerTokenAuth,
     Collection,
     ExternalDocs,
     FormItem,
@@ -70,23 +74,58 @@ def extract_server_variables(spec: dict[str, Any]) -> dict[str, dict[str, str]]:
             "description": f"Server URL {i+1}: {server.get('description', '')}",
         }
 
-    # # Extract security schemes
-    # security_schemes = spec.get("components", {}).get("securitySchemes", {})
-    # for scheme_name, scheme in security_schemes.items():
-    #     if scheme["type"] == "apiKey":
-    #         variables[f"{scheme_name.upper()}_API_KEY"] = {
-    #             "value": "YOUR_API_KEY_HERE",
-    #             "description": f"API Key for {scheme_name} authentication",
-    #         }
-    #     elif scheme["type"] == "http" and scheme["scheme"] == "bearer":
-    #         variables[f"{scheme_name.upper()}_BEARER_TOKEN"] = {
-    #             "value": "YOUR_BEARER_TOKEN_HERE",
-    #             "description": f"Bearer token for {scheme_name} authentication",
-    #         }
-
     return variables
 
 
+def security_scheme_to_variables(
+    name: str,
+    security_scheme: SecurityScheme | Reference,
+) -> dict[str, dict[str, str]]:
+    match security_scheme:
+        case SecurityScheme(type="http", scheme="basic"):
+            return {
+                f"{name.upper()}_USERNAME": {
+                    "value": "YOUR USERNAME HERE",
+                    "description": f"Username for {name} authentication",
+                },
+                f"{name.upper()}_PASSWORD": {
+                    "value": "YOUR PASSWORD HERE",
+                    "description": f"Password for {name} authentication",
+                },
+            }
+        case SecurityScheme(type="http", scheme="bearer"):
+            return {
+                f"{name.upper()}_BEARER_TOKEN": {
+                    "value": "YOUR BEARER TOKEN HERE",
+                    "description": f"Token for {name} authentication",
+                },
+            }
+        case _:
+            return {}
+
+
+def security_scheme_to_auth(
+    name: str,
+    security_scheme: SecurityScheme | Reference,
+) -> Auth | None:
+    match security_scheme:
+        case SecurityScheme(type="http", scheme="basic"):
+            return Auth(
+                type="basic",
+                basic=BasicAuth(
+                    username=f"${{{name.upper()}_USERNAME}}",
+                    password=f"${{{name.upper()}_PASSWORD}}",
+                ),
+            )
+        case SecurityScheme(type="http", scheme="bearer"):
+            return Auth(
+                type="bearer_token",
+                bearer_token=BearerTokenAuth(token=f"${{{name.upper()}_BEARER_TOKEN}}"),
+            )
+        case _:
+            return None
+
+
 def generate_readme(
     spec_path: Path,
     info: APIInfo,
@@ -184,9 +223,16 @@ def import_openapi_spec(spec_path: str | Path) -> Collection:
         name=collection_name,
     )
 
+    openapi = OpenAPI.model_validate(spec)
+    security_schemes = openapi.components.securitySchemes or {}
+
     env_files: list[Path] = []
     for server in servers:
-        variables = extract_server_variables(server)
+        security_variables = {}
+        for scheme_name, scheme in security_schemes.items():
+            security_variables.update(security_scheme_to_variables(scheme_name, scheme))
+
+        variables = {**extract_server_variables(server), **security_variables}
         env_filename = generate_unique_env_filename(collection_name, server["url"])
         env_file = create_env_file(spec_path.parent, env_filename, variables)
         console.print(
@@ -209,6 +255,14 @@ def import_openapi_spec(spec_path: str | Path) -> Collection:
                 method=method,
                 url=f"${{BASE_URL}}{path}",
             )
+
+            # Add auth
+            for security in operation.get("security", []):
+                for scheme_name, _scopes in security.items():
+                    if scheme := security_schemes.get(scheme_name):
+                        request.auth = security_scheme_to_auth(scheme_name, scheme)
+                        break
+
             # Add query parameters
             for param in operation.get("parameters", []):
                 if param["in"] == "query":

src/posting/widgets/request/request_auth.py

@@ -6,7 +6,8 @@
 from textual.containers import Horizontal, Vertical, VerticalScroll
 from textual.widgets import ContentSwitcher, Input, Label, Select, Static
 
-from posting.collection import Auth, BasicAuth, DigestAuth
+from posting.auth import HttpxBearerTokenAuth
+from posting.collection import Auth, BasicAuth, BearerTokenAuth, DigestAuth
 from posting.widgets.select import PostingSelect
 from posting.widgets.variable_input import VariableInput
 
@@ -52,6 +53,34 @@ def get_values(self) -> dict[str, str]:
         }
 
 
+class BearerTokenForm(Vertical):
+    DEFAULT_CSS = """
+    BearerTokenForm {
+        padding: 1 0;
+
+        & #token-input {
+            margin-bottom: 1;
+        }
+    }
+    """
+
+    def compose(self) -> ComposeResult:
+        yield Label("Token")
+        yield VariableInput(
+            placeholder="Enter a token",
+            password=True,
+            id="token-input",
+        )
+
+    def set_values(self, token: str) -> None:
+        self.query_one("#token-input", Input).value = token
+
+    def get_values(self) -> dict[str, str]:
+        return {
+            "token": self.query_one("#token-input", Input).value,
+        }
+
+
 class RequestAuth(VerticalScroll):
     DEFAULT_CSS = """
     RequestAuth {
@@ -93,6 +122,7 @@ def compose(self) -> ComposeResult:
                         ("No Auth", None),
                         ("Basic", "basic"),
                         ("Digest", "digest"),
+                        ("Bearer Token", "bearer-token"),
                     ],
                     allow_blank=False,
                     prompt="Auth Type",
@@ -107,6 +137,7 @@ def compose(self) -> ComposeResult:
         with ContentSwitcher(initial=None, id="auth-form-switcher"):
             yield UserNamePasswordForm(id="auth-form-basic")
             yield UserNamePasswordForm(id="auth-form-digest")
+            yield BearerTokenForm(id="auth-form-bearer-token")
 
     @on(Select.Changed, selector="#auth-type-select")
     def on_auth_type_changed(self, event: Select.Changed):
@@ -123,6 +154,8 @@ def to_httpx_auth(self) -> httpx.Auth | None:
                 return httpx.BasicAuth(**form.get_values())
             case "auth-form-digest":
                 return httpx.DigestAuth(**form.get_values())
+            case "auth-form-bearer-token":
+                return HttpxBearerTokenAuth(**form.get_values())
             case _:
                 return None
 
@@ -147,6 +180,10 @@ def to_model(self) -> Auth | None:
                     type="digest",
                     digest=DigestAuth(username=username, password=password),
                 )
+            case "auth-form-bearer-token":
+                form_values = form.get_values()
+                token = form_values["token"]
+                return Auth(type="bearer_token", bearer_token=BearerTokenAuth(token=token))
             case _:
                 return None
 
@@ -177,6 +214,14 @@ def load_auth(self, auth: Auth | None) -> None:
                     auth.digest.username,
                     auth.digest.password,
                 )
+            case "bearer_token":
+                if auth.bearer_token is None:
+                    log.warning("Bearer auth selected, but no values provided for token.")
+                    return
+                self.query_one("#auth-type-select", Select).value = "bearer-token"
+                self.query_one("#auth-form-bearer-token", BearerTokenForm).set_values(
+                    auth.bearer_token.token
+                )
             case _:
                 log.warning(f"Unknown auth type: {auth.type}")
 

tests/test_curl_import.py

@@ -82,6 +82,14 @@ def test_curl_with_user_and_password():
     assert curl_import.url == "http://example.com"
 
 
+def test_curl_with_bearer_token():
+    """Test parsing of user credentials."""
+    curl_command = "curl http://example.com -H 'Authorization: Bearer my-token'"
+    curl_import = CurlImport(curl_command)
+    assert curl_import.headers == [("Authorization", "Bearer my-token")]
+    assert curl_import.url == "http://example.com"
+
+
 def test_curl_with_insecure():
     """Test parsing of --insecure flag."""
     curl_command = "curl -k http://example.com"