# Multi-Tenant Connectors — Implementation Plan
Created: June 30, 2026
Completed: June 30, 2026
Purpose: Complete the multi-tenant connector stack — tier enforcement, credential encryption, admin UX
## Status: ✅ COMPLETE
All three tasks implemented and tested:
- **Task 1: Tier Enforcement** — 12/12 tests passing
- **Task 2: Credential Encryption** — All access points migrated, migration script ready
- **Task 3: Admin UX** — Build succeeded, 4 UX fixes applied
---
## Current State Audit
**What's already done:**
- `Connector` model has `company_id` FK — data isolation at DB level ✓
- All connector routes filter by `_get_company_id()` — tenant isolation ✓
- CRM data (contacts/deals) scoped by `company_id` ✓
- Analytics queries filter by tenant ✓
- Scheduler runs per-connector, scoped to company ✓
- Plans endpoint returns `limits.integrations` per tier ✓
**Tier limits (from `api_proxy.py`):**
| Tier | Integrations |
|------|-------------|
| Starter | 2 |
| Growth | 5 |
| Command | 99 |
| Enterprise | 99 |
---
## Gap 1: Tier Enforcement
**Problem:** Connector creation does not check plan limits. A Starter user can create unlimited connectors.
### 1.1 Enforce connector count on creation
**File:** `app/routes/connectors.py` — `POST /api/connectors`
**Current code (line ~120):**
```python
@connectors_bp.route('/api/connectors', methods=['POST'])
@login_required
def create_connector():
company_id = _get_company_id()
data = request.get_json()
service = data.get('service')
# ... creates connector without checking limits
```
**Change:** Add tier limit check before creating connector.
```python
@connectors_bp.route('/api/connectors', methods=['POST'])
@login_required
def create_connector():
company_id = _get_company_id()
data = request.get_json()
service = data.get('service')
# Check tier limit
company = Company.query.get(company_id)
current_count = Connector.query.filter_by(company_id=company_id).count()
tier = company.settings_json.get('tier', 'starter') # or subscription_tier field
limits = {
'starter': 2,
'growth': 5,
'command': 99,
'enterprise': 99,
}
max_connectors = limits.get(tier, 2)
if current_count >= max_connectors:
return jsonify({
'error': f'Connector limit reached ({current_count}/{max_connectors}). Upgrade your plan to add more.'
}), 403
# ... proceed with creation
```
**Also check:** `app/routes/api_proxy.py` — plans endpoint currently returns hardcoded `'tier': 'growth'` (line ~363). Make this dynamic based on actual `Company` subscription.
### 1.2 Sync frequency gating
**Problem:** Launch/Growth tiers should not receive real-time sync. Only Command/Enterprise.
**File:** `app/routes/connectors.py` — `PUT /api/connectors/<id>`
**Change:** When updating sync_frequency, validate against tier.
```python
# In update_connector route
if sync_frequency == 'real_time' and tier in ('starter', 'growth'):
return jsonify({
'error': 'Real-time sync requires Command or Enterprise plan.'
}), 403
```
### 1.3 API: Return tier info with connector list
**File:** `app/routes/connectors.py` — `GET /api/connectors`
**Change:** Add `tier`, `connector_limit`, `connectors_used` to response.
```python
response = {
'connectors': [...],
'tier': company.settings_json.get('tier', 'starter'),
'connector_limit': max_connectors,
'connectors_used': current_count,
}
```
---
## Gap 2: Credential Encryption
**Problem:** `Connector.config_json` stores API keys, OAuth tokens, and secrets as plain JSON in the database.
### 2.1 Encryption utility
**File:** `app/utils/encryption.py` (new)
```python
import os
from cryptography.fernet import Fernet
# Use a key from environment — generate once, store securely
_KEY = os.environ.get('CONNECTOR_ENCRYPTION_KEY')
if _KEY:
_fernet = Fernet(_KEY.encode() if len(_KEY) == 44 else base64.urlsafe_b64encode(_KEY.encode()))
else:
# Fallback: generate ephemeral key (NOT secure for production)
_fernet = Fernet(Fernet.generate_key())
def encrypt_config(config: dict) -> str:
"""Encrypt a config dict to Fernet token string."""
import json
plaintext = json.dumps(config).encode()
return _fernet.encrypt(plaintext).decode()
def decrypt_config(token: str) -> dict:
"""Decrypt a Fernet token string back to config dict."""
import json
plaintext = _fernet.decrypt(token.encode())
return json.loads(plaintext)
```
**Env var:** `CONNECTOR_ENCRYPTION_KEY` — Fernet key (44 bytes). Generate with:
```bash
python3 -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
```
### 2.2 Update Connector model
**File:** `app/models.py` — `Connector` class
**Current:** `config_json = db.Column(db.JSON)`
**Change:** Add hybrid properties for encrypted storage.
```python
class Connector(db.Model):
id = db.Column(db.String(36), primary_key=True)
company_id = db.Column(db.String(36), db.ForeignKey('company.id'), nullable=False)
service = db.Column(db.String(50), nullable=False)
config_encrypted = db.Column(db.Text, nullable=True) # NEW: encrypted config
config_json = db.Column(db.JSON, nullable=True) # DEPRECATED: plaintext (keep for migration)
@property
def config(self):
"""Read config — decrypt if encrypted, fallback to plaintext."""
if self.config_encrypted:
from app.utils.encryption import decrypt_config
return decrypt_config(self.config_encrypted)
return self.config_json or {}
@config.setter
def config(self, value: dict):
"""Write config — always encrypt."""
from app.utils.encryption import encrypt_config
self.config_encrypted = encrypt_config(value)
self.config_json = None # Clear plaintext
```
### 2.3 Migrate existing connectors
**File:** `app/utils/migrate_encryption.py` (new one-time script)
```python
"""Run once to encrypt existing plaintext config_json."""
from app import create_app, db
from app.models import Connector
from app.utils.encryption import encrypt_config
app = create_app()
with app.app_context():
connectors = Connector.query.filter(Connector.config_json.isnot(None)).all()
for c in connectors:
if c.config_json:
c.config_encrypted = encrypt_config(c.config_json)
c.config_json = None
db.session.commit()
print(f"Encrypted {len(connectors)} connectors")
```
**Run:** `cd /home/vincent/projects/command-sovereignty && python3 app/utils/migrate_encryption.py`
### 2.4 Update all config access points
**Files to update:**
- `app/routes/connectors.py` — `create_connector()` — encrypt on write
- `app/connectors/base.py` — `_validate_config()` — decrypt for validation
- `app/scheduler.py` — `_run_connector()` — decrypt before passing to connector class
- `frontend/src/api/connectors.ts` — never expose encrypted config to frontend (already redacted server-side ✓)
**Pattern:** Always use `connector.config` property (hybrid) — never read `config_encrypted` directly.
---
## Gap 3: Admin UX
**File:** `frontend/src/pages/admin/AdminIntegrationsPage.tsx`
### 3.1 Fix sync frequency dropdown
**Current (line 177):** `onChange={(e) => {}}`
**Change:** Wire to API call.
```typescript
const handleSyncFrequency = async (connectorId: string, frequency: string) => {
try {
await connectorsApi.update(connectorId, { sync_frequency: frequency });
setToast({ message: 'Sync frequency updated', type: 'success' });
await fetchData();
} catch (err: any) {
setToast({ message: err.response?.data?.error || 'Update failed', type: 'error' });
}
};
// In AdminActiveCard:
onChange={(e) => handleSyncFrequency(connector.id, e.target.value)}
```
**API change needed:** `app/routes/connectors.py` — `PUT /api/connectors/<id>` must accept `sync_frequency` in update payload.
### 3.2 Show connector count vs limit
**Current (line 477-492):** Shows connected/available/active/error counts — no tier limit.
**Change:** Add tier badge showing usage.
```typescript
// Fetch tier info from GET /api/connectors response
const [tierInfo, setTierInfo] = useState<{ tier: string; limit: number; used: number } | null>(null);
// In header:
{tierInfo && (
<div className="flex items-center gap-2 mt-2">
<span className={`rounded-full px-2 py-0.5 text-[10px] font-medium ${
tierInfo.used >= tierInfo.limit
? 'bg-red-50 text-red-600'
: tierInfo.used >= tierInfo.limit * 0.8
? 'bg-yellow-50 text-yellow-600'
: 'bg-green-50 text-green-600'
}`}>
{tierInfo.used}/{tierInfo.limit} connectors ({tierInfo.tier} plan)
</span>
</div>
)}
```
### 3.3 Show real connector stats
**Current (line 197-213):** Hardcoded `-` for contacts, deals, syncs.
**Change:** Fetch from analytics API.
```typescript
// Add to connector record interface:
interface ConnectorRecord {
id: string;
service: string;
status: string;
contacts_count?: number;
deals_count?: number;
sync_count?: number;
// ...
}
// In AdminActiveCard stats section:
<p className="text-lg font-bold text-surface-900">
{connector.contacts_count ?? 0}
</p>
<p className="text-lg font-bold text-surface-900">
{connector.deals_count ?? 0}
</p>
<p className="text-lg font-bold text-surface-900">
{connector.sync_count ?? 0}
</p>
```
**API change needed:** `app/routes/connectors.py` — `GET /api/connectors` must include counts.
```python
# In list_connectors response:
connector_dict = {
'id': c.id,
'service': c.service,
'status': c.status,
'contacts_count': CrmContact.query.filter_by(company_id=company_id, source_service=c.service).count(),
'deals_count': CrmDeal.query.filter_by(company_id=company_id, source_service=c.service).count(),
'sync_count': SyncLog.query.filter_by(connector_id=c.id).count(),
# ...
}
```
### 3.4 Upgrade prompt when limit reached
**When:** User clicks "Connect" on available connector but has hit their limit.
**Change:** Show modal instead of connect form.
```typescript
// In handleConnect:
const handleConnect = (connector: ConnectorDef) => {
if (tierInfo && tierInfo.used >= tierInfo.limit) {
setShowUpgradeModal(true);
return;
}
setConnectModal(connector);
};
// Upgrade modal:
{showUpgradeModal && (
<div className="fixed inset-0 z-[60] flex items-center justify-center bg-black/50 p-4">
<div className="w-full max-w-md rounded-2xl bg-white p-6 shadow-xl">
<h2 className="text-lg font-semibold mb-2">Connector Limit Reached</h2>
<p className="text-sm text-surface-500 mb-4">
Your {tierInfo?.tier} plan includes {tierInfo?.limit} connectors.
Upgrade to add more integrations.
</p>
<div className="flex gap-3">
<button onClick={() => setShowUpgradeModal(false)} className="flex-1 ...">
Cancel
</button>
<a href="/admin/billing" className="flex-1 ...">
Upgrade Plan
</a>
</div>
</div>
</div>
)}
```
---
## Implementation Order
1. **Tier enforcement** (backend) — `app/routes/connectors.py`, `app/routes/api_proxy.py`
2. **Credential encryption** — `app/utils/encryption.py`, `app/models.py`, `app/scheduler.py`, migration script
3. **Admin UX** — `frontend/src/pages/admin/AdminIntegrationsPage.tsx`, API changes for stats
---
## Delegation Tasks
### Task 1: Tier Enforcement (Backend)
**Goal:** Add plan limit checks to connector creation and update. Gate real-time sync by tier.
**Files:** `app/routes/connectors.py`, `app/routes/api_proxy.py`
**Deliverables:**
- `POST /api/connectors` checks tier limit before creating
- `PUT /api/connectors/<id>` validates sync_frequency against tier
- `GET /api/connectors` returns tier info + usage count
- `api_proxy.py` plans endpoint returns dynamic tier from `Company.settings_json`
- Tests for limit enforcement
### Task 2: Credential Encryption
**Goal:** Encrypt `config_json` at rest using Fernet symmetric encryption.
**Files:** `app/utils/encryption.py` (new), `app/models.py`, `app/scheduler.py`, `app/connectors/base.py`
**Deliverables:**
- Encryption utility with env var key
- Hybrid property on Connector model (encrypt on write, decrypt on read)
- Migration script for existing plaintext configs
- All config access points updated to use hybrid property
- Tests for encrypt/decrypt round-trip
### Task 3: Admin UX
**Goal:** Fix sync dropdown, show tier usage badge, display real connector stats.
**Files:** `frontend/src/pages/admin/AdminIntegrationsPage.tsx`, `app/routes/connectors.py` (stats), `frontend/src/api/connectors.ts` (types)
**Deliverables:**
- Sync frequency dropdown wired to API
- Tier usage badge in header
- Real contact/deal/sync counts per connector
- Upgrade modal when limit reached
- TypeScript types updated
---
## Testing Strategy
1. **Tier enforcement:** Create connector as Starter user → 3rd attempt returns 403
2. **Sync gating:** Growth tier attempts real_time sync → returns 403
3. **Encryption:** Create connector → DB shows encrypted blob → `connector.config` returns original dict
4. **Migration:** Run migration script → existing plaintext configs become encrypted
5. **Admin UX:** Verify badge counts, dropdown works, stats populate
---
## Risk Assessment
| Risk | Likelihood | Mitigation |
|------|-----------|------------|
| Encryption key lost = data loss | Low | Document key generation, store in secrets manager |
| Migration script fails mid-way | Low | Wrap in transaction, test on copy first |
| Tier field not set on existing companies | Medium | Default to 'starter' if missing |
| Real-time sync breaking on tier downgrade | Low | Scheduler validates tier before running |
---
## Notes
- **NO schema migration needed** — `config_json` column stays (deprecated), new `config_encrypted` column added. Soft migration: new connectors use encrypted, old connectors migrated via script.
- **Encryption key** should be generated once and stored in environment. Never in code or git.
- **Tier field** — currently in `Company.settings_json`. Consider adding `subscription_tier` column for query performance if tier enforcement grows.
- **Delegation:** Each task is independent — can be parallelized to 3 Claude subagents.