maciejpi/nordabiz
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(oauth): Phase 3 foundation - OAuth 2.0 framework for external APIs
Some checks are pending
NordaBiz Tests / Unit & Integration Tests (push) Waiting to run
Details
NordaBiz Tests / E2E Tests (Playwright) (push) Blocked by required conditions
Details
NordaBiz Tests / Smoke Tests (Production) (push) Blocked by required conditions
Details
NordaBiz Tests / Send Failure Notification (push) Blocked by required conditions
Details

Browse Source 
New files:
- oauth_service.py: Shared OAuth 2.0 service supporting Google and Meta
  providers with token exchange, refresh, and storage
- database/migrations/058_oauth_tokens.sql: oauth_tokens table with
  company/provider/service unique constraint
- blueprints/api/routes_oauth.py: OAuth API endpoints for connect,
  callback, status, and disconnect flows

Supports:
- Google OAuth (GBP Business Profile, Search Console)
- Meta OAuth (Facebook Pages, Instagram)
- CSRF state validation, token refresh, expiry tracking
- Per-company token storage with active/inactive status

Requires .env config:
- GOOGLE_OAUTH_CLIENT_ID, GOOGLE_OAUTH_CLIENT_SECRET (Google APIs)
- META_APP_ID, META_APP_SECRET (Facebook/Instagram)
- OAUTH_REDIRECT_BASE_URL (default: https://nordabiznes.pl)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Maciej Pienczyn 2026-02-08 11:46:42 +01:00
parent 279947d4aa
commit 66cd223568
6 changed files with 598 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
blueprints/api/__init__.py
View File

@ -18,3 +18,4 @@ from . import routes_social_audit # noqa: E402, F401
from . import routes_company # noqa: E402, F401
from . import routes_membership # noqa: E402, F401
from . import routes_audit_actions # noqa: E402, F401
from . import routes_oauth # noqa: E402, F401

184
blueprints/api/routes_oauth.py Normal file
View File

@ -0,0 +1,184 @@
"""
OAuth API Routes
================
Endpoints for connecting external accounts (Google, Meta) via OAuth 2.0.
"""
import logging
import secrets
from flask import jsonify, request, redirect, session
from flask_login import login_required, current_user
from . import bp
from database import SessionLocal, OAuthToken
logger = logging.getLogger(__name__)
@bp.route('/oauth/connect/<provider>/<service>', methods=['POST'])
@login_required
def oauth_connect(provider, service):
"""Initiate OAuth flow for connecting an external account.
POST /api/oauth/connect/google/gbp
POST /api/oauth/connect/meta/facebook
"""
from oauth_service import OAuthService
# Validate provider/service
valid_combinations = {
('google', 'gbp'), ('google', 'search_console'),
('meta', 'facebook'), ('meta', 'instagram'),
}
if (provider, service) not in valid_combinations:
return jsonify({'success': False, 'error': f'Nieznany provider/service: {provider}/{service}'}), 400
# User must have a company
if not current_user.company_id:
return jsonify({'success': False, 'error': 'Musisz być przypisany do firmy'}), 403
# Generate CSRF state token
state = f"{current_user.company_id}:{current_user.id}:{service}:{secrets.token_urlsafe(16)}"
session['oauth_state'] = state
oauth = OAuthService()
auth_url = oauth.get_authorization_url(provider, service, state)
if not auth_url:
return jsonify({
'success': False,
'error': f'OAuth nie skonfigurowany dla {provider}. Skontaktuj się z administratorem.'
}), 503
return jsonify({'success': True, 'auth_url': auth_url})
@bp.route('/oauth/callback/<provider>', methods=['GET'])
@login_required
def oauth_callback(provider):
"""Handle OAuth callback from provider.
GET /api/oauth/callback/google?code=...&state=...
GET /api/oauth/callback/meta?code=...&state=...
"""
from oauth_service import OAuthService
code = request.args.get('code')
state = request.args.get('state')
error = request.args.get('error')
if error:
logger.warning(f"OAuth error from {provider}: {error}")
return redirect(f'/admin/company?oauth_error={error}')
if not code or not state:
return redirect('/admin/company?oauth_error=missing_params')
# Validate state
saved_state = session.pop('oauth_state', None)
if not saved_state or saved_state != state:
logger.warning(f"OAuth state mismatch for {provider}")
return redirect('/admin/company?oauth_error=invalid_state')
# Parse state: company_id:user_id:service:random
try:
parts = state.split(':')
company_id = int(parts[0])
user_id = int(parts[1])
service = parts[2]
except (ValueError, IndexError):
return redirect('/admin/company?oauth_error=invalid_state_format')
# Verify user owns this company
if current_user.id != user_id or current_user.company_id != company_id:
return redirect('/admin/company?oauth_error=unauthorized')
# Exchange code for token
oauth = OAuthService()
token_data = oauth.exchange_code(provider, code)
if not token_data:
return redirect('/admin/company?oauth_error=token_exchange_failed')
# Save token
db = SessionLocal()
try:
success = oauth.save_token(db, company_id, user_id, provider, service, token_data)
if success:
logger.info(f"OAuth connected: {provider}/{service} for company {company_id} by user {user_id}")
return redirect(f'/admin/company?oauth_success={provider}/{service}')
else:
return redirect('/admin/company?oauth_error=save_failed')
finally:
db.close()
@bp.route('/oauth/status', methods=['GET'])
@login_required
def oauth_status():
"""Get connected OAuth services for current user's company.
GET /api/oauth/status
"""
from oauth_service import OAuthService
if not current_user.company_id:
return jsonify({'success': True, 'services': {}})
db = SessionLocal()
try:
oauth = OAuthService()
services = oauth.get_connected_services(db, current_user.company_id)
# Add available (but not connected) services
all_services = {
'google/gbp': {'name': 'Google Business Profile', 'description': 'Pełne dane o wizytówce, opinie, insights'},
'google/search_console': {'name': 'Google Search Console', 'description': 'Zapytania, CTR, pozycje w wyszukiwaniu'},
'meta/facebook': {'name': 'Facebook', 'description': 'Reach, impressions, demographics, post insights'},
'meta/instagram': {'name': 'Instagram', 'description': 'Stories, reels, engagement metrics'},
}
result = {}
for key, info in all_services.items():
connected = services.get(key, {})
result[key] = {
**info,
'connected': bool(connected),
'account_name': connected.get('account_name'),
'expires_at': connected.get('expires_at'),
'is_expired': connected.get('is_expired', False),
}
return jsonify({'success': True, 'services': result})
finally:
db.close()
@bp.route('/oauth/disconnect/<provider>/<service>', methods=['POST'])
@login_required
def oauth_disconnect(provider, service):
"""Disconnect an OAuth service.
POST /api/oauth/disconnect/google/gbp
"""
if not current_user.company_id:
return jsonify({'success': False, 'error': 'Brak przypisanej firmy'}), 403
db = SessionLocal()
try:
token = db.query(OAuthToken).filter(
OAuthToken.company_id == current_user.company_id,
OAuthToken.provider == provider,
OAuthToken.service == service,
).first()
if token:
token.is_active = False
db.commit()
logger.info(f"OAuth disconnected: {provider}/{service} for company {current_user.company_id}")
return jsonify({'success': True, 'message': f'{provider}/{service} rozłączony'})
else:
return jsonify({'success': False, 'error': 'Nie znaleziono połączenia'}), 404
finally:
db.close()

37
database.py
View File

@ -5182,6 +5182,43 @@ class SocialConnection(Base):
return f"<SocialConnection {self.id} company={self.company_id} platform={self.platform}>"
class OAuthToken(Base):
"""OAuth tokens for external API integrations (Google, Meta)."""
__tablename__ = 'oauth_tokens'
id = Column(Integer, primary_key=True)
company_id = Column(Integer, ForeignKey('companies.id'), nullable=False)
company = relationship('Company', backref='oauth_tokens')
user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
user = relationship('User', backref='oauth_tokens')
provider = Column(String(50), nullable=False) # google, meta
service = Column(String(50), nullable=False) # gbp, search_console, facebook, instagram
access_token = Column(Text, nullable=False)
refresh_token = Column(Text)
token_type = Column(String(50), default='Bearer')
expires_at = Column(DateTime)
scopes = Column(Text)
account_id = Column(String(255))
account_name = Column(String(255))
metadata_json = Column('metadata', PG_JSONB)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.now)
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
__table_args__ = (
UniqueConstraint('company_id', 'provider', 'service', name='uq_oauth_company_provider_service'),
)
@property
def is_expired(self):
if not self.expires_at:
return False
return datetime.now() > self.expires_at
def __repr__(self):
return f'<OAuthToken {self.provider}/{self.service} for company_id={self.company_id}>'
# ============================================================
# DATABASE INITIALIZATION
# ============================================================

29
database/migrations/058_oauth_tokens.sql Normal file
View File

@ -0,0 +1,29 @@
-- OAuth Tokens for external API integrations
-- Supports: Google (GBP Business Profile, Search Console), Meta (Facebook, Instagram)
CREATE TABLE IF NOT EXISTS oauth_tokens (
id SERIAL PRIMARY KEY,
company_id INTEGER NOT NULL REFERENCES companies(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL REFERENCES users(id),
provider VARCHAR(50) NOT NULL, -- 'google', 'meta'
service VARCHAR(50) NOT NULL, -- 'gbp', 'search_console', 'facebook', 'instagram'
access_token TEXT NOT NULL,
refresh_token TEXT,
token_type VARCHAR(50) DEFAULT 'Bearer',
expires_at TIMESTAMP,
scopes TEXT, -- space-separated scopes
account_id VARCHAR(255), -- external account/page ID
account_name VARCHAR(255), -- external account/page name
metadata JSONB, -- additional provider-specific data
is_active BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW(),
UNIQUE(company_id, provider, service)
);
CREATE INDEX idx_oauth_tokens_company ON oauth_tokens(company_id);
CREATE INDEX idx_oauth_tokens_provider ON oauth_tokens(provider, service);
-- Grant permissions
GRANT ALL ON TABLE oauth_tokens TO nordabiz_app;
GRANT USAGE, SELECT ON SEQUENCE oauth_tokens_id_seq TO nordabiz_app;

30
docs/AUDIT_COMPLETENESS_PLAN.md
View File

@ -40,20 +40,20 @@
- [ ] Migracja bazy danych (nowe kolumny JSONB — opcjonalne, dane w result dict)
- [ ] Zaktualizować szablony HTML (wyświetlanie atrybutów)
### Faza 3: OAuth Framework (0 PLN API, 2-4 tygodnie dev)
- [ ] Shared OAuth 2.0 framework (`oauth_service.py`)
### Faza 3: OAuth Framework (0 PLN API, 2-4 tygodnie dev) — FUNDAMENT UKOŃCZONY (2026-02-08)
- [x] Shared OAuth 2.0 framework (`oauth_service.py`) — Google + Meta providers
- [x] Tabela `oauth_tokens` w DB (migracja 058)
- [x] Model `OAuthToken` w database.py
- [x] API endpoints: `/api/oauth/connect`, `/api/oauth/callback`, `/api/oauth/status`, `/api/oauth/disconnect`
- [ ] GBP Business Profile API:
- Scope: `business.manage`, App review ~14 dni, darmowe
- Daje: WSZYSTKIE opinie (nie max 5), owner responses, insights (views/clicks/calls/keywords), posty
- Wymaga: GOOGLE_OAUTH_CLIENT_ID, GOOGLE_OAUTH_CLIENT_SECRET w .env
- Daje: WSZYSTKIE opinie, owner responses, insights, posty
- [ ] Facebook + Instagram Graph API:
- Wspólny OAuth via Meta, App review 3-7 dni
- Scopes: pages_show_list, pages_read_engagement, read_insights, instagram_basic, instagram_manage_insights
- Wymaga: META_APP_ID, META_APP_SECRET w .env + App review 3-7 dni
- Daje: reach, impressions, demographics, post insights, IG stories/reels
- Token: Long-Lived (90 dni), Page Token (nigdy nie wygasa)
- [ ] Google Search Console API (per firma OAuth, darmowe)
- Daje: zapytania wyszukiwania, CTR, pozycje, status indeksacji
- [ ] UI: "Połącz konto" w panelu firmy
- [ ] Tabela `oauth_tokens` w DB
- [ ] UI: "Połącz konto" w panelu firmy (frontend)
### Faza 4: Zaawansowane (opcjonalne)
- [ ] Sentiment analysis recenzji via Gemini
@ -98,9 +98,11 @@
## Wpływ na Kompletność
| | Obecny | F0 | F1 | F2 | F3 |
| | Początkowy | F0 ✅ | F1 ✅ | F2 ✅ | F3 (plan) |
|---|--------|-----|-----|-----|-----|
| GBP | 55% | 60% | 75% | 90% | 98% |
| SEO | 60% | 75% | 85% | 85% | 95% |
| Social | 35% | 50% | 65% | 65% | 85% |
| **Średnia** | **52%** | **68%** | **78%** | **83%** | **93%** |
| GBP | 55% | 60% | 75% | **90%** | 98% |
| SEO | 60% | 75% | **85%** | 85% | 95% |
| Social | 35% | 50% | **65%** | 65% | 85% |
| **Średnia** | **52%** | **68%** | **78%** | **~83%** | **93%** |
**Status (2026-02-08):** F0+F1+F2 ukończone. Obecna kompletność: ~83%. Pozostała: F3 (OAuth).

331
oauth_service.py Normal file
View File

@ -0,0 +1,331 @@
"""
Shared OAuth 2.0 Service for NordaBiz
=====================================
Supports:
- Google OAuth 2.0 (GBP Business Profile API, Search Console API)
- Meta OAuth 2.0 (Facebook Graph API, Instagram Graph API)
Config via environment variables:
- GOOGLE_OAUTH_CLIENT_ID, GOOGLE_OAUTH_CLIENT_SECRET
- META_APP_ID, META_APP_SECRET
- OAUTH_REDIRECT_BASE_URL (e.g., https://nordabiznes.pl)
"""
import os
import logging
from datetime import datetime, timedelta
from typing import Optional, Dict, Tuple
import requests
logger = logging.getLogger(__name__)
# OAuth Provider Configurations
OAUTH_PROVIDERS = {
'google': {
'auth_url': 'https://accounts.google.com/o/oauth2/v2/auth',
'token_url': 'https://oauth2.googleapis.com/token',
'scopes': {
'gbp': 'https://www.googleapis.com/auth/business.manage',
'search_console': 'https://www.googleapis.com/auth/webmasters.readonly',
},
},
'meta': {
'auth_url': 'https://www.facebook.com/v21.0/dialog/oauth',
'token_url': 'https://graph.facebook.com/v21.0/oauth/access_token',
'scopes': {
'facebook': 'pages_show_list,pages_read_engagement,read_insights',
'instagram': 'instagram_basic,instagram_manage_insights,pages_show_list',
},
},
}
class OAuthService:
"""Shared OAuth 2.0 service for external API integrations."""
def __init__(self):
self.google_client_id = os.getenv('GOOGLE_OAUTH_CLIENT_ID')
self.google_client_secret = os.getenv('GOOGLE_OAUTH_CLIENT_SECRET')
self.meta_app_id = os.getenv('META_APP_ID')
self.meta_app_secret = os.getenv('META_APP_SECRET')
self.redirect_base = os.getenv('OAUTH_REDIRECT_BASE_URL', 'https://nordabiznes.pl')
def get_authorization_url(self, provider: str, service: str, state: str) -> Optional[str]:
"""Generate OAuth authorization URL for a provider/service.
Args:
provider: 'google' or 'meta'
service: 'gbp', 'search_console', 'facebook', 'instagram'
state: CSRF state token (include company_id and user_id encoded)
Returns:
Authorization URL or None if provider not configured
"""
config = OAUTH_PROVIDERS.get(provider)
if not config:
logger.error(f"Unknown OAuth provider: {provider}")
return None
scopes = config['scopes'].get(service, '')
redirect_uri = f"{self.redirect_base}/api/oauth/callback/{provider}"
if provider == 'google':
if not self.google_client_id:
logger.warning("Google OAuth not configured (GOOGLE_OAUTH_CLIENT_ID)")
return None
params = {
'client_id': self.google_client_id,
'redirect_uri': redirect_uri,
'response_type': 'code',
'scope': scopes,
'state': state,
'access_type': 'offline',
'prompt': 'consent',
}
elif provider == 'meta':
if not self.meta_app_id:
logger.warning("Meta OAuth not configured (META_APP_ID)")
return None
params = {
'client_id': self.meta_app_id,
'redirect_uri': redirect_uri,
'response_type': 'code',
'scope': scopes,
'state': state,
}
else:
return None
query = '&'.join(f'{k}={requests.utils.quote(str(v))}' for k, v in params.items())
return f"{config['auth_url']}?{query}"
def exchange_code(self, provider: str, code: str) -> Optional[Dict]:
"""Exchange authorization code for access token.
Args:
provider: 'google' or 'meta'
code: Authorization code from callback
Returns:
Dict with access_token, refresh_token, expires_in or None on error
"""
config = OAUTH_PROVIDERS.get(provider)
if not config:
return None
redirect_uri = f"{self.redirect_base}/api/oauth/callback/{provider}"
if provider == 'google':
data = {
'code': code,
'client_id': self.google_client_id,
'client_secret': self.google_client_secret,
'redirect_uri': redirect_uri,
'grant_type': 'authorization_code',
}
elif provider == 'meta':
data = {
'code': code,
'client_id': self.meta_app_id,
'client_secret': self.meta_app_secret,
'redirect_uri': redirect_uri,
}
else:
return None
try:
response = requests.post(config['token_url'], data=data, timeout=15)
if response.status_code != 200:
logger.error(f"OAuth token exchange failed for {provider}: {response.status_code} - {response.text}")
return None
return response.json()
except Exception as e:
logger.error(f"OAuth token exchange error for {provider}: {e}")
return None
def refresh_access_token(self, provider: str, refresh_token: str) -> Optional[Dict]:
"""Refresh an expired access token.
Args:
provider: 'google' or 'meta'
refresh_token: The refresh token
Returns:
Dict with new access_token, expires_in or None
"""
if provider == 'google':
data = {
'client_id': self.google_client_id,
'client_secret': self.google_client_secret,
'refresh_token': refresh_token,
'grant_type': 'refresh_token',
}
token_url = OAUTH_PROVIDERS['google']['token_url']
elif provider == 'meta':
# Meta long-lived tokens: exchange short-lived for long-lived
params = {
'grant_type': 'fb_exchange_token',
'client_id': self.meta_app_id,
'client_secret': self.meta_app_secret,
'fb_exchange_token': refresh_token,
}
try:
response = requests.get(
OAUTH_PROVIDERS['meta']['token_url'],
params=params,
timeout=15
)
if response.status_code == 200:
return response.json()
return None
except Exception as e:
logger.error(f"Meta token refresh error: {e}")
return None
else:
return None
try:
response = requests.post(token_url, data=data, timeout=15)
if response.status_code == 200:
return response.json()
logger.error(f"Token refresh failed for {provider}: {response.status_code}")
return None
except Exception as e:
logger.error(f"Token refresh error for {provider}: {e}")
return None
def save_token(self, db, company_id: int, user_id: int, provider: str,
service: str, token_data: Dict) -> bool:
"""Save or update OAuth token in database.
Args:
db: Database session
company_id: Company ID
user_id: User who authorized
provider: 'google' or 'meta'
service: 'gbp', 'search_console', 'facebook', 'instagram'
token_data: Dict from exchange_code() or refresh_access_token()
Returns:
True if saved successfully
"""
try:
from database import OAuthToken
# Find existing token or create new
token = db.query(OAuthToken).filter(
OAuthToken.company_id == company_id,
OAuthToken.provider == provider,
OAuthToken.service == service,
).first()
if not token:
token = OAuthToken(
company_id=company_id,
user_id=user_id,
provider=provider,
service=service,
)
db.add(token)
token.access_token = token_data.get('access_token', '')
token.refresh_token = token_data.get('refresh_token', token.refresh_token)
token.token_type = token_data.get('token_type', 'Bearer')
expires_in = token_data.get('expires_in')
if expires_in:
token.expires_at = datetime.now() + timedelta(seconds=int(expires_in))
token.scopes = OAUTH_PROVIDERS.get(provider, {}).get('scopes', {}).get(service, '')
token.is_active = True
token.updated_at = datetime.now()
db.commit()
logger.info(f"OAuth token saved: {provider}/{service} for company {company_id}")
return True
except Exception as e:
db.rollback()
logger.error(f"Error saving OAuth token: {e}")
return False
def get_valid_token(self, db, company_id: int, provider: str, service: str) -> Optional[str]:
"""Get a valid access token, refreshing if needed.
Args:
db: Database session
company_id: Company ID
provider: 'google' or 'meta'
service: Service name
Returns:
Valid access token string or None
"""
try:
from database import OAuthToken
token = db.query(OAuthToken).filter(
OAuthToken.company_id == company_id,
OAuthToken.provider == provider,
OAuthToken.service == service,
OAuthToken.is_active == True,
).first()
if not token:
return None
# Check if token is expired
if token.is_expired and token.refresh_token:
new_data = self.refresh_access_token(provider, token.refresh_token)
if new_data:
token.access_token = new_data.get('access_token', token.access_token)
expires_in = new_data.get('expires_in')
if expires_in:
token.expires_at = datetime.now() + timedelta(seconds=int(expires_in))
token.updated_at = datetime.now()
db.commit()
else:
# Refresh failed, mark as inactive
token.is_active = False
db.commit()
return None
return token.access_token
except Exception as e:
logger.error(f"Error getting valid token: {e}")
return None
def get_connected_services(self, db, company_id: int) -> Dict[str, Dict]:
"""Get status of all connected OAuth services for a company.
Returns:
Dict like {'google/gbp': {'connected': True, 'account_name': '...', 'expires_at': ...}, ...}
"""
try:
from database import OAuthToken
tokens = db.query(OAuthToken).filter(
OAuthToken.company_id == company_id,
OAuthToken.is_active == True,
).all()
result = {}
for token in tokens:
key = f"{token.provider}/{token.service}"
result[key] = {
'connected': True,
'account_name': token.account_name,
'account_id': token.account_id,
'expires_at': token.expires_at.isoformat() if token.expires_at else None,
'is_expired': token.is_expired,
'updated_at': token.updated_at.isoformat() if token.updated_at else None,
}
return result
except Exception as e:
logger.error(f"Error getting connected services: {e}")
return {}