Merge pull request #552 from tfranzel/swagger_non_public

Swagger UI authorized schema retrieval #342 #458

drf_spectacular/templates/drf_spectacular/swagger_ui.js

@@ -1,18 +1,118 @@
 "use strict";
 
-const swagger_settings = {{ settings|safe }};
+const swaggerSettings = {{ settings|safe }};
+const schemaAuthNames = {{ schema_auth_names|safe }};
+let schemaAuthFailed = false;
+const plugins = [];
+
+const reloadSchemaOnAuthChange = () => {
+  return {
+    statePlugins: {
+      auth: {
+        wrapActions: {
+          authorize: (ori) => (...args) => {
+            schemaAuthFailed = false;
+            setTimeout(() => ui.specActions.download());
+            return ori(...args);
+          },
+          logout: (ori) => (...args) => {
+            schemaAuthFailed = false;
+            setTimeout(() => ui.specActions.download());
+            return ori(...args);
+          },
+        },
+      },
+    },
+  };
+};
+
+if (schemaAuthNames.length > 0) {
+  plugins.push(reloadSchemaOnAuthChange);
+}
+
+const uiInitialized = () => {
+  try {
+    ui;
+    return true;
+  } catch {
+    return false;
+  }
+};
+
+const isSchemaUrl = (url) => {
+  if (!uiInitialized()) {
+    return false;
+  }
+  return url === new URL(ui.getConfigs().url, document.baseURI).href;
+};
+
+const responseInterceptor = (response, ...args) => {
+  if (!response.ok && isSchemaUrl(response.url)) {
+    console.warn("schema request received '" + response.status + "'. disabling credentials for schema till logout.");
+    if (!schemaAuthFailed) {
+      // only retry once to prevent endless loop.
+      schemaAuthFailed = true;
+      setTimeout(() => ui.specActions.download());
+    }
+  }
+  return response;
+};
+
+const injectAuthCredentials = (request) => {
+  let authorized;
+  if (uiInitialized()) {
+    const state = ui.getState().get("auth").get("authorized");
+    if (state !== undefined && Object.keys(state.toJS()).length !== 0) {
+      authorized = state.toJS();
+    }
+  } else if (![undefined, "{}"].includes(localStorage.authorized)) {
+    authorized = JSON.parse(localStorage.authorized);
+  }
+  if (authorized === undefined) {
+    return;
+  }
+  for (const authName of schemaAuthNames) {
+    const authDef = authorized[authName];
+    if (authDef === undefined || authDef.schema === undefined) {
+      continue;
+    }
+    if (authDef.schema.type === "http" && authDef.schema.scheme === "bearer") {
+      request.headers["Authorization"] = "Bearer " + authDef.value;
+      return;
+    } else if (authDef.schema.type === "http" && authDef.schema.scheme === "basic") {
+      request.headers["Authorization"] = "Basic " + btoa(authDef.value.username + ":" + authDef.value.password);
+      return;
+    } else if (authDef.schema.type === "apiKey" && authDef.schema.in === "header") {
+      request.headers[authDef.schema.name] = authDef.value;
+      return;
+    }
+  }
+};
+
+const requestInterceptor = (request, ...args) => {
+  if (request.loadSpec && schemaAuthNames.length > 0 && !schemaAuthFailed) {
+    try {
+      injectAuthCredentials(request);
+    } catch (e) {
+      console.error("schema auth injection failed with error: ", e);
+    }
+  }
+  // selectively omit adding headers to mitigate CORS issues.
+  if (!["GET", undefined].includes(request.method) && request.credentials === "same-origin") {
+    request.headers["{{ csrf_header_name }}"] = "{{ csrf_token }}";
+  }
+  return request;
+};
 
 const ui = SwaggerUIBundle({
   url: "{{ schema_url }}",
   dom_id: "#swagger-ui",
   presets: [SwaggerUIBundle.presets.apis],
-  plugin: [SwaggerUIBundle.plugins.DownloadUrl],
+  plugins,
   layout: "BaseLayout",
-  requestInterceptor: (request) => {
-    request.headers["X-CSRFToken"] = "{{ csrf_token }}";
-    return request;
-  },
-  ...swagger_settings,
+  requestInterceptor,
+  responseInterceptor,
+  ...swaggerSettings,
 });
 
 {% if oauth2_config %}ui.initOAuth({{ oauth2_config|safe }});{% endif %}

drf_spectacular/views.py

@@ -116,13 +116,31 @@ def get(self, request, *args, **kwargs):
                 ),
                 'settings': self._dump(spectacular_settings.SWAGGER_UI_SETTINGS),
                 'oauth2_config': self._dump(spectacular_settings.SWAGGER_UI_OAUTH2_CONFIG),
-                'template_name_js': self.template_name_js
+                'template_name_js': self.template_name_js,
+                'csrf_header_name': self._get_csrf_header_name(),
+                'schema_auth_names': self._dump(self._get_schema_auth_names()),
             },
             template_name=self.template_name,
         )
 
     def _dump(self, data):
-        return data if isinstance(data, str) else json.dumps(data)
+        return data if isinstance(data, str) else json.dumps(data, indent=2)
+
+    def _get_csrf_header_name(self):
+        csrf_header_name = settings.CSRF_HEADER_NAME
+        if csrf_header_name.startswith('HTTP_'):
+            csrf_header_name = csrf_header_name[5:]
+        return csrf_header_name.replace('_', '-')
+
+    def _get_schema_auth_names(self):
+        from drf_spectacular.extensions import OpenApiAuthenticationExtension
+        if spectacular_settings.SERVE_PUBLIC:
+            return []
+        auth_extensions = [
+            OpenApiAuthenticationExtension.get_match(klass)
+            for klass in self.authentication_classes
+        ]
+        return [auth.name for auth in auth_extensions if auth]
 
     def _swagger_ui_dist(self):
         if spectacular_settings.SWAGGER_UI_DIST == 'SIDECAR':
@@ -154,6 +172,8 @@ def get(self, request, *args, **kwargs):
                     ),
                     'settings': self._dump(spectacular_settings.SWAGGER_UI_SETTINGS),
                     'oauth2_config': self._dump(spectacular_settings.SWAGGER_UI_OAUTH2_CONFIG),
+                    'csrf_header_name': self._get_csrf_header_name(),
+                    'schema_auth_names': self._dump(self._get_schema_auth_names()),
                 },
                 template_name=self.template_name_js,
                 content_type='application/javascript',

tests/test_view.py

@@ -131,4 +131,4 @@ def test_spectacular_swagger_ui_alternate(no_warnings):
 def test_spectacular_ui_with_raw_settings(no_warnings):
     response = APIClient().get('/api/v2/schema/swagger-ui/')
     assert response.status_code == 200
-    assert b'const swagger_settings = {"deepLinking": true};\n\n' in response.content
+    assert b'const swaggerSettings = {"deepLinking": true};\n' in response.content