274 lines
13 KiB
Go
274 lines
13 KiB
Go
|
package saas
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"crypto/sha256"
|
||
|
|
"encoding/hex"
|
||
|
|
"encoding/json"
|
||
|
|
"errors"
|
||
|
|
"fmt"
|
||
|
|
"strings"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"data-terminal-cloud/internal/audit"
|
||
|
|
"data-terminal-cloud/internal/database"
|
||
|
|
"data-terminal-cloud/internal/identity"
|
||
|
|
"data-terminal-cloud/internal/outbox"
|
||
|
|
|
||
|
|
"github.com/jackc/pgx/v5"
|
||
|
|
)
|
||
|
|
|
||
|
|
var ErrIdempotencyConflict = errors.New("idempotency key was reused with different request data")
|
||
|
|
|
||
|
|
type PrivacySubtask struct {
|
||
|
|
ID string `json:"id"`
|
||
|
|
Subsystem string `json:"subsystem"`
|
||
|
|
TargetType string `json:"target_type"`
|
||
|
|
TargetRef string `json:"target_ref"`
|
||
|
|
Status string `json:"status"`
|
||
|
|
Attempts int `json:"attempts"`
|
||
|
|
NextAttemptAt *time.Time `json:"next_attempt_at,omitempty"`
|
||
|
|
LastError string `json:"last_error,omitempty"`
|
||
|
|
CompletedAt *time.Time `json:"completed_at,omitempty"`
|
||
|
|
}
|
||
|
|
|
||
|
|
type PrivacyRequest struct {
|
||
|
|
ID string `json:"id"`
|
||
|
|
SubjectID string `json:"subject_id"`
|
||
|
|
RequestType string `json:"request_type"`
|
||
|
|
Scope map[string]any `json:"scope"`
|
||
|
|
Status string `json:"status"`
|
||
|
|
DeadlineAt *time.Time `json:"deadline_at,omitempty"`
|
||
|
|
LegalHold bool `json:"legal_hold"`
|
||
|
|
ResultSummary map[string]any `json:"result_summary"`
|
||
|
|
Version int64 `json:"version"`
|
||
|
|
CreatedAt time.Time `json:"created_at"`
|
||
|
|
UpdatedAt time.Time `json:"updated_at"`
|
||
|
|
CompletedAt *time.Time `json:"completed_at,omitempty"`
|
||
|
|
Subtasks []PrivacySubtask `json:"subtasks"`
|
||
|
|
}
|
||
|
|
|
||
|
|
type CreatePrivacyRequestInput struct {
|
||
|
|
RequestType string `json:"request_type"`
|
||
|
|
Scope map[string]any `json:"scope"`
|
||
|
|
Reason string `json:"-"`
|
||
|
|
Confirmation string `json:"-"`
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) CreatePrivacyRequest(ctx context.Context, principal identity.Principal, subjectPublicID, idempotencyKey, requestID string, input CreatePrivacyRequestInput) (PrivacyRequest, bool, error) {
|
||
|
|
input.RequestType = strings.ToUpper(strings.TrimSpace(input.RequestType))
|
||
|
|
input.Reason = strings.TrimSpace(input.Reason)
|
||
|
|
input.Confirmation = strings.TrimSpace(input.Confirmation)
|
||
|
|
if input.Scope == nil {
|
||
|
|
input.Scope = map[string]any{}
|
||
|
|
}
|
||
|
|
if input.RequestType != "ACCESS" && input.RequestType != "DELETE" {
|
||
|
|
return PrivacyRequest{}, false, ErrNotFound
|
||
|
|
}
|
||
|
|
if input.RequestType == "DELETE" {
|
||
|
|
reasonDigest := sha256.Sum256([]byte(input.Reason))
|
||
|
|
input.Scope = map[string]any{"reason_digest": hex.EncodeToString(reasonDigest[:])}
|
||
|
|
}
|
||
|
|
confirmationDigest := ""
|
||
|
|
if input.RequestType == "DELETE" {
|
||
|
|
value := sha256.Sum256([]byte(input.Confirmation))
|
||
|
|
confirmationDigest = hex.EncodeToString(value[:])
|
||
|
|
}
|
||
|
|
requestBytes, err := json.Marshal(struct {
|
||
|
|
RequestType string `json:"request_type"`
|
||
|
|
Scope map[string]any `json:"scope"`
|
||
|
|
ConfirmationDigest string `json:"confirmation_digest,omitempty"`
|
||
|
|
}{input.RequestType, input.Scope, confirmationDigest})
|
||
|
|
if err != nil {
|
||
|
|
return PrivacyRequest{}, false, fmt.Errorf("encode privacy request: %w", err)
|
||
|
|
}
|
||
|
|
digestValue := sha256.Sum256(requestBytes)
|
||
|
|
digest := hex.EncodeToString(digestValue[:])
|
||
|
|
created := false
|
||
|
|
var publicID string
|
||
|
|
err = database.Transact(ctx, s.pool, func(tx pgx.Tx) error {
|
||
|
|
// Idempotent replay remains readable after the first DELETE request has
|
||
|
|
// frozen the subject and removed it from normal resource resolution.
|
||
|
|
var replayDigest string
|
||
|
|
replayErr := tx.QueryRow(ctx, `
|
||
|
|
SELECT pr.public_id::text,pr.request_digest
|
||
|
|
FROM privacy_request pr
|
||
|
|
JOIN subject_profile sp ON sp.id=pr.subject_id AND sp.tenant_id=pr.tenant_id
|
||
|
|
JOIN household_member hm ON hm.household_id=sp.household_id AND hm.tenant_id=sp.tenant_id AND hm.status='active'
|
||
|
|
JOIN user_account ua ON ua.id=hm.user_id AND ua.tenant_id=sp.tenant_id AND ua.status='active'
|
||
|
|
WHERE sp.public_id=$1::uuid AND pr.idempotency_key=$2
|
||
|
|
AND ua.oidc_issuer=$3 AND ua.oidc_subject=$4 AND pr.requested_by=ua.id`,
|
||
|
|
subjectPublicID, idempotencyKey, principal.Issuer, principal.Subject).Scan(&publicID, &replayDigest)
|
||
|
|
if replayErr == nil {
|
||
|
|
if replayDigest != digest {
|
||
|
|
return ErrIdempotencyConflict
|
||
|
|
}
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
if !errors.Is(replayErr, pgx.ErrNoRows) {
|
||
|
|
return replayErr
|
||
|
|
}
|
||
|
|
|
||
|
|
access, subject, err := resolveSubject(ctx, tx, principal, subjectPublicID, identity.CapabilityPrivacyManage, true)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if input.RequestType == "DELETE" {
|
||
|
|
if input.Confirmation != subject.PreferredName {
|
||
|
|
return ErrDeleteConfirmation
|
||
|
|
}
|
||
|
|
var activeCount int
|
||
|
|
if err := tx.QueryRow(ctx, `SELECT count(*) FROM conversation WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND status NOT IN ('COMPLETED','FAILED','EXPIRED')`, access.tenantID, access.subjectID).Scan(&activeCount); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if activeCount > 0 {
|
||
|
|
return ErrSubjectDeleteConflict
|
||
|
|
}
|
||
|
|
}
|
||
|
|
scopeJSON, err := json.Marshal(input.Scope)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
var internalID string
|
||
|
|
err = tx.QueryRow(ctx, `
|
||
|
|
INSERT INTO privacy_request(tenant_id,subject_id,request_type,scope,requested_by,idempotency_key,request_digest)
|
||
|
|
VALUES($1::uuid,$2::uuid,$3,$4::jsonb,$5::uuid,$6,$7)
|
||
|
|
ON CONFLICT (tenant_id,requested_by,idempotency_key) DO NOTHING
|
||
|
|
RETURNING id::text,public_id::text`, access.tenantID, access.subjectID, input.RequestType, scopeJSON, access.userID, idempotencyKey, digest,
|
||
|
|
).Scan(&internalID, &publicID)
|
||
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
||
|
|
var storedDigest string
|
||
|
|
if err := tx.QueryRow(ctx, `SELECT public_id::text,request_digest FROM privacy_request WHERE tenant_id=$1::uuid AND requested_by=$2::uuid AND idempotency_key=$3 FOR UPDATE`, access.tenantID, access.userID, idempotencyKey).Scan(&publicID, &storedDigest); err != nil {
|
||
|
|
return fmt.Errorf("read idempotent privacy request: %w", err)
|
||
|
|
}
|
||
|
|
if storedDigest != digest {
|
||
|
|
return ErrIdempotencyConflict
|
||
|
|
}
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return fmt.Errorf("create privacy request: %w", err)
|
||
|
|
}
|
||
|
|
created = true
|
||
|
|
eventType := "privacy.access_requested"
|
||
|
|
action := "privacy.access.request"
|
||
|
|
if input.RequestType == "ACCESS" {
|
||
|
|
if _, err := tx.Exec(ctx, `INSERT INTO privacy_subtask(privacy_request_id,subsystem,target_type,target_ref) VALUES($1::uuid,'postgresql','subject_data_inventory',$2)`, internalID, subjectPublicID); err != nil {
|
||
|
|
return fmt.Errorf("create privacy subtask: %w", err)
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
tasks := [][2]string{{"redis", "ephemeral_keys"}, {"export_files", "private_exports"}, {"object_storage", "subject_media"}, {"pgvector", "memory_embeddings"}, {"provider_index", "external_memory_index"}, {"postgresql", "subject_personal_data"}}
|
||
|
|
for _, task := range tasks {
|
||
|
|
if _, err := tx.Exec(ctx, `INSERT INTO privacy_subtask(privacy_request_id,subsystem,target_type,target_ref) VALUES($1::uuid,$2,$3,$4)`, internalID, task[0], task[1], subjectPublicID); err != nil {
|
||
|
|
return fmt.Errorf("create deletion subtask %s: %w", task[0], err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
if _, err := tx.Exec(ctx, `UPDATE subject_profile SET status='deleting',profile_version=profile_version+1,updated_by=$3::uuid,updated_at=now() WHERE tenant_id=$1::uuid AND id=$2::uuid`, access.tenantID, access.subjectID, access.userID); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if _, err := tx.Exec(ctx, `UPDATE consent_record SET withdrawn_at=COALESCE(withdrawn_at,GREATEST(now(),granted_at)),scope='{}'::jsonb,evidence_ref=NULL,version=version+1,updated_at=now() WHERE tenant_id=$1::uuid AND subject_id=$2::uuid`, access.tenantID, access.subjectID); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if _, err := tx.Exec(ctx, `UPDATE curated_memory SET status='DISABLED',valid_to=COALESCE(valid_to,now()),version=version+1,updated_at=now() WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND status NOT IN ('DELETED','DISABLED')`, access.tenantID, access.subjectID); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
if _, err := tx.Exec(ctx, `UPDATE memory_index_entry SET embedding=NULL,status='STALE',last_error='subject privacy deletion requested',updated_at=now() WHERE tenant_id=$1::uuid AND subject_id=$2::uuid AND status<>'DELETED'`, access.tenantID, access.subjectID); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
eventType = "privacy.delete_requested"
|
||
|
|
action = "privacy.delete.request"
|
||
|
|
}
|
||
|
|
dedupe := strings.ReplaceAll(eventType, ".", "-") + ":" + internalID
|
||
|
|
payload, _ := json.Marshal(map[string]any{"request_id": internalID})
|
||
|
|
if _, err := outbox.NewStore(s.pool).Enqueue(ctx, tx, outbox.NewEvent{TenantID: &access.tenantID, AggregateType: "privacy_request", AggregateID: internalID, EventType: eventType, DeduplicationKey: &dedupe, Payload: payload}); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
return s.audit.Record(ctx, tx, audit.Event{TenantID: strPtr(access.tenantID), ActorType: "user", ActorPublicID: strPtr(access.userPublicID), Action: action, ResourceType: "privacy_request", ResourceID: strPtr(publicID), RequestID: requestID, Result: "success", Detail: map[string]any{"subject_id": subjectPublicID, "scope": input.Scope}})
|
||
|
|
})
|
||
|
|
if err != nil {
|
||
|
|
return PrivacyRequest{}, false, err
|
||
|
|
}
|
||
|
|
result, err := s.GetPrivacyRequest(ctx, principal, publicID)
|
||
|
|
return result, created, err
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) ListPrivacyRequests(ctx context.Context, principal identity.Principal, subjectPublicID string) ([]PrivacyRequest, error) {
|
||
|
|
access, _, err := resolveSubject(ctx, s.pool, principal, subjectPublicID, identity.CapabilityPrivacyRead, false)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
rows, err := s.pool.Query(ctx, privacyRequestSelectSQL+` WHERE pr.tenant_id=$1::uuid AND pr.subject_id=$2::uuid ORDER BY pr.created_at DESC,pr.id DESC`, access.tenantID, access.subjectID)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
defer rows.Close()
|
||
|
|
items := make([]PrivacyRequest, 0)
|
||
|
|
for rows.Next() {
|
||
|
|
item, err := scanPrivacyRequest(rows)
|
||
|
|
if err != nil {
|
||
|
|
return nil, err
|
||
|
|
}
|
||
|
|
items = append(items, item)
|
||
|
|
}
|
||
|
|
return items, rows.Err()
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) GetPrivacyRequest(ctx context.Context, principal identity.Principal, publicID string) (PrivacyRequest, error) {
|
||
|
|
item, err := scanPrivacyRequest(s.pool.QueryRow(ctx, privacyRequestSelectSQL+`
|
||
|
|
JOIN household h ON h.id=sp.household_id AND h.tenant_id=sp.tenant_id AND h.status<>'deleted'
|
||
|
|
JOIN household_member hm ON hm.household_id=h.id AND hm.tenant_id=h.tenant_id AND hm.status='active'
|
||
|
|
JOIN user_account ua ON ua.id=hm.user_id AND ua.tenant_id=h.tenant_id AND ua.status='active'
|
||
|
|
WHERE pr.public_id=$1::uuid AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`, publicID, principal.Issuer, principal.Subject))
|
||
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
||
|
|
return PrivacyRequest{}, ErrAccessDenied
|
||
|
|
}
|
||
|
|
if err != nil {
|
||
|
|
return PrivacyRequest{}, err
|
||
|
|
}
|
||
|
|
var role string
|
||
|
|
if err := s.pool.QueryRow(ctx, `SELECT hm.role FROM privacy_request pr JOIN subject_profile sp ON sp.id=pr.subject_id JOIN household_member hm ON hm.household_id=sp.household_id AND hm.status='active' JOIN user_account ua ON ua.id=hm.user_id AND ua.tenant_id=pr.tenant_id AND ua.status='active' WHERE pr.public_id=$1::uuid AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`, publicID, principal.Issuer, principal.Subject).Scan(&role); err != nil {
|
||
|
|
return PrivacyRequest{}, ErrAccessDenied
|
||
|
|
}
|
||
|
|
if !identity.Can(identity.Role(role), identity.CapabilityPrivacyRead) {
|
||
|
|
return PrivacyRequest{}, ErrAccessDenied
|
||
|
|
}
|
||
|
|
if err := s.loadPrivacySubtasks(ctx, &item); err != nil {
|
||
|
|
return PrivacyRequest{}, err
|
||
|
|
}
|
||
|
|
return item, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
const privacyRequestSelectSQL = `SELECT pr.public_id::text,sp.public_id::text,pr.request_type,pr.scope,pr.status,pr.deadline_at,pr.legal_hold,pr.result_summary,pr.version,pr.created_at,pr.updated_at,pr.completed_at FROM privacy_request pr JOIN subject_profile sp ON sp.id=pr.subject_id AND sp.tenant_id=pr.tenant_id`
|
||
|
|
|
||
|
|
func scanPrivacyRequest(row rowScanner) (PrivacyRequest, error) {
|
||
|
|
var item PrivacyRequest
|
||
|
|
var scopeJSON, resultJSON []byte
|
||
|
|
if err := row.Scan(&item.ID, &item.SubjectID, &item.RequestType, &scopeJSON, &item.Status, &item.DeadlineAt, &item.LegalHold, &resultJSON, &item.Version, &item.CreatedAt, &item.UpdatedAt, &item.CompletedAt); err != nil {
|
||
|
|
return PrivacyRequest{}, err
|
||
|
|
}
|
||
|
|
if err := decodeObject(scopeJSON, &item.Scope); err != nil {
|
||
|
|
return PrivacyRequest{}, err
|
||
|
|
}
|
||
|
|
if err := decodeObject(resultJSON, &item.ResultSummary); err != nil {
|
||
|
|
return PrivacyRequest{}, err
|
||
|
|
}
|
||
|
|
item.Subtasks = []PrivacySubtask{}
|
||
|
|
return item, nil
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *Service) loadPrivacySubtasks(ctx context.Context, item *PrivacyRequest) error {
|
||
|
|
rows, err := s.pool.Query(ctx, `SELECT ps.public_id::text,ps.subsystem,ps.target_type,ps.target_ref,ps.status,ps.attempts,ps.next_attempt_at,COALESCE(ps.last_error,''),ps.completed_at FROM privacy_subtask ps JOIN privacy_request pr ON pr.id=ps.privacy_request_id WHERE pr.public_id=$1::uuid ORDER BY ps.created_at,ps.id`, item.ID)
|
||
|
|
if err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
defer rows.Close()
|
||
|
|
for rows.Next() {
|
||
|
|
var subtask PrivacySubtask
|
||
|
|
if err := rows.Scan(&subtask.ID, &subtask.Subsystem, &subtask.TargetType, &subtask.TargetRef, &subtask.Status, &subtask.Attempts, &subtask.NextAttemptAt, &subtask.LastError, &subtask.CompletedAt); err != nil {
|
||
|
|
return err
|
||
|
|
}
|
||
|
|
item.Subtasks = append(item.Subtasks, subtask)
|
||
|
|
}
|
||
|
|
return rows.Err()
|
||
|
|
}
|