kunthawat/ALwrity
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Harden OAuth callback postMessage origin and payload encoding

Browse Source
This commit is contained in:
ي
2026-05-11 15:47:59 +05:30
committed by ajaysi
parent 8834a05cf5
commit 11d83e6f86
3 changed files with 173 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

218
backend/routers/wordpress_oauth.py
View File

@@ -8,6 +8,9 @@ from fastapi.responses import RedirectResponse, HTMLResponse, JSONResponse
from typing import Dict, Any, Optional
from pydantic import BaseModel
from loguru import logger
import json
import os
from urllib.parse import urlparse
from services.integrations.wordpress_oauth import WordPressOAuthService
from middleware.auth_middleware import get_current_user
@@ -17,6 +20,65 @@ router = APIRouter(prefix="/wp", tags=["WordPress OAuth"])
# Initialize OAuth service
oauth_service = WordPressOAuthService()
def _sanitize_string(value: Any, max_len: int = 500) -> str:
if value is None:
return ""
return " ".join(str(value).split())[:max_len]
def _normalize_origin(url: Optional[str]) -> Optional[str]:
if not url:
return None
parsed = urlparse(url.strip())
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
return None
return f"{parsed.scheme}://{parsed.netloc}"
def _trusted_frontend_origin() -> Optional[str]:
origins_env = os.getenv("OAUTH_CALLBACK_ALLOWED_ORIGINS", "")
configured_origins = [
_normalize_origin(origin)
for origin in origins_env.split(",")
if origin.strip()
]
configured_origins = [origin for origin in configured_origins if origin]
if configured_origins:
return configured_origins[0]
return _normalize_origin(os.getenv("FRONTEND_URL"))
def _oauth_callback_html(payload: Dict[str, Any], title: str, heading: str, message: str) -> str:
payload_json = json.dumps(payload)
target_origin = json.dumps(_trusted_frontend_origin() or "")
heading_html = heading.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
message_html = message.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
return f"""
<!DOCTYPE html>
<html>
<head><title>{title}</title></head>
<body>
<h1>{heading_html}</h1>
<p>{message_html}</p>
<script>
(function() {{
var payload = {payload_json};
var targetOrigin = {target_origin};
var destination = window.opener || window.parent;
if (destination && targetOrigin) {{
try {{
destination.postMessage(payload, targetOrigin);
window.close();
return;
}} catch (_e) {{}}
}}
}})();
</script>
</body>
</html>
"""
# Pydantic Models
class WordPressOAuthResponse(BaseModel):
auth_url: str
@@ -78,30 +140,12 @@ async def handle_wordpress_callback(
status_code=status.HTTP_400_BAD_REQUEST,
content={"success": False, "error": error}
)
html_content = f"""
<!DOCTYPE html>
<html>
<head>
<title>WordPress.com Connection Failed</title>
<script>
// Send error message to parent window
window.onload = function() {{
(window.opener || window.parent).postMessage({{
type: 'WPCOM_OAUTH_ERROR',
success: false,
error: '{error}'
}}, '*');
window.close();
}};
</script>
</head>
<body>
<h1>Connection Failed</h1>
<p>There was an error connecting to WordPress.com.</p>
<p>You can close this window and try again.</p>
</body>
</html>
"""
html_content = _oauth_callback_html(
payload={"type": "WPCOM_OAUTH_ERROR", "success": False, "error": _sanitize_string(error)},
title="WordPress.com Connection Failed",
heading="Connection Failed",
message="There was an error connecting to WordPress.com. You can close this window and try again."
)
return HTMLResponse(content=html_content, headers={
"Cross-Origin-Opener-Policy": "unsafe-none",
"Cross-Origin-Embedder-Policy": "unsafe-none"
@@ -114,30 +158,12 @@ async def handle_wordpress_callback(
status_code=status.HTTP_400_BAD_REQUEST,
content={"success": False, "error": "Missing parameters"}
)
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>WordPress.com Connection Failed</title>
<script>
// Send error message to opener/parent window
window.onload = function() {{
(window.opener || window.parent).postMessage({{
type: 'WPCOM_OAUTH_ERROR',
success: false,
error: 'Missing parameters'
}}, '*');
window.close();
}};
</script>
</head>
<body>
<h1>Connection Failed</h1>
<p>Missing required parameters.</p>
<p>You can close this window and try again.</p>
</body>
</html>
"""
html_content = _oauth_callback_html(
payload={"type": "WPCOM_OAUTH_ERROR", "success": False, "error": "Missing parameters"},
title="WordPress.com Connection Failed",
heading="Connection Failed",
message="Missing required parameters. You can close this window and try again."
)
return HTMLResponse(content=html_content, headers={
"Cross-Origin-Opener-Policy": "unsafe-none",
"Cross-Origin-Embedder-Policy": "unsafe-none"
@@ -153,30 +179,12 @@ async def handle_wordpress_callback(
status_code=status.HTTP_400_BAD_REQUEST,
content={"success": False, "error": "Token exchange failed"}
)
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>WordPress.com Connection Failed</title>
<script>
// Send error message to opener/parent window
window.onload = function() {{
(window.opener || window.parent).postMessage({{
type: 'WPCOM_OAUTH_ERROR',
success: false,
error: 'Token exchange failed'
}}, '*');
window.close();
}};
</script>
</head>
<body>
<h1>Connection Failed</h1>
<p>Failed to exchange authorization code for access token.</p>
<p>You can close this window and try again.</p>
</body>
</html>
"""
html_content = _oauth_callback_html(
payload={"type": "WPCOM_OAUTH_ERROR", "success": False, "error": "Token exchange failed"},
title="WordPress.com Connection Failed",
heading="Connection Failed",
message="Failed to exchange authorization code for access token. You can close this window and try again."
)
return HTMLResponse(content=html_content)
# Return success page with postMessage script
@@ -193,31 +201,17 @@ async def handle_wordpress_callback(
}
)
html_content = f"""
<!DOCTYPE html>
<html>
<head>
<title>WordPress.com Connection Successful</title>
<script>
// Send success message to opener/parent window
window.onload = function() {{
(window.opener || window.parent).postMessage({{
type: 'WPCOM_OAUTH_SUCCESS',
success: true,
blogUrl: '{blog_url}',
blogId: '{blog_id}'
}}, '*');
window.close();
}};
</script>
</head>
<body>
<h1>Connection Successful!</h1>
<p>Your WordPress.com site has been connected successfully.</p>
<p>You can close this window now.</p>
</body>
</html>
"""
html_content = _oauth_callback_html(
payload={
"type": "WPCOM_OAUTH_SUCCESS",
"success": True,
"blogUrl": _sanitize_string(blog_url, 300),
"blogId": _sanitize_string(blog_id, 128)
},
title="WordPress.com Connection Successful",
heading="Connection Successful",
message="Your WordPress.com site has been connected successfully. You can close this window now."
)
return HTMLResponse(content=html_content, headers={
"Cross-Origin-Opener-Policy": "unsafe-none",
@@ -226,30 +220,12 @@ async def handle_wordpress_callback(
except Exception as e:
logger.error(f"Error handling WordPress OAuth callback: {e}")
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>WordPress.com Connection Failed</title>
<script>
// Send error message to opener/parent window
window.onload = function() {{
(window.opener || window.parent).postMessage({{
type: 'WPCOM_OAUTH_ERROR',
success: false,
error: 'Callback error'
}}, '*');
window.close();
}};
</script>
</head>
<body>
<h1>Connection Failed</h1>
<p>An unexpected error occurred during connection.</p>
<p>You can close this window and try again.</p>
</body>
</html>
"""
html_content = _oauth_callback_html(
payload={"type": "WPCOM_OAUTH_ERROR", "success": False, "error": "Callback error"},
title="WordPress.com Connection Failed",
heading="Connection Failed",
message="An unexpected error occurred during connection. You can close this window and try again."
)
return HTMLResponse(content=html_content, headers={
"Cross-Origin-Opener-Policy": "unsafe-none",
"Cross-Origin-Embedder-Policy": "unsafe-none"