Codex Local
bf1233d5df
ci / verify (push) Canceled after 0s
Add customer and operator login, BLE provisioning, device fleet controls, call memory, memoir generation, and refreshed family and maintenance portals. Include API contracts, migrations, firmware sources, tests, assets, and deployment documentation.
570 lines
31 KiB
Go
570 lines
31 KiB
Go
package sharing
|
|
|
|
import (
|
|
"context"
|
|
"crypto/hmac"
|
|
"crypto/sha256"
|
|
"crypto/subtle"
|
|
"encoding/base64"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"data-terminal-cloud/internal/audit"
|
|
"data-terminal-cloud/internal/idempotency"
|
|
"data-terminal-cloud/internal/identity"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
var (
|
|
ErrAccessDenied = errors.New("sharing access denied")
|
|
ErrInvalid = errors.New("sharing grant is invalid")
|
|
ErrNotFound = errors.New("sharing resource not found")
|
|
ErrExpired = errors.New("sharing grant expired or revoked")
|
|
ErrVersionConflict = errors.New("shared resource version conflict")
|
|
)
|
|
|
|
type mediaSigner interface {
|
|
PresignMediaDownload(context.Context, string, time.Duration, string, string, bool) (string, error)
|
|
}
|
|
|
|
type Service struct {
|
|
pool *pgxpool.Pool
|
|
audit *audit.Repository
|
|
idem *idempotency.Executor
|
|
store mediaSigner
|
|
accessTTL time.Duration
|
|
tokenSecret []byte
|
|
}
|
|
|
|
type CreateInput struct {
|
|
ResourceType string `json:"resource_type"`
|
|
ResourceID string `json:"resource_id"`
|
|
GranteeUserID string `json:"grantee_user_id"`
|
|
Permission string `json:"permission"`
|
|
Purpose string `json:"purpose"`
|
|
ExpiresAt time.Time `json:"expires_at"`
|
|
LoginRequired bool `json:"login_required"`
|
|
}
|
|
|
|
type Grant struct {
|
|
ID string `json:"id"`
|
|
SubjectID string `json:"subject_id"`
|
|
ResourceType string `json:"resource_type"`
|
|
ResourceID string `json:"resource_id"`
|
|
GranteeUserID string `json:"grantee_user_id"`
|
|
Permission string `json:"permission"`
|
|
Purpose string `json:"purpose"`
|
|
LoginRequired bool `json:"login_required"`
|
|
GrantedBy string `json:"granted_by"`
|
|
GrantedAt time.Time `json:"granted_at"`
|
|
ExpiresAt time.Time `json:"expires_at"`
|
|
RevokedAt *time.Time `json:"revoked_at,omitempty"`
|
|
Version int64 `json:"version"`
|
|
AccessToken string `json:"access_token,omitempty"`
|
|
}
|
|
|
|
type SharedTurn struct {
|
|
SequenceNo int64 `json:"sequence_no"`
|
|
Speaker string `json:"speaker"`
|
|
StartAt time.Time `json:"start_at"`
|
|
EndAt *time.Time `json:"end_at,omitempty"`
|
|
Text string `json:"text"`
|
|
Interrupted bool `json:"interrupted"`
|
|
}
|
|
|
|
type SharedResource struct {
|
|
GrantID string `json:"grant_id"`
|
|
ResourceType string `json:"resource_type"`
|
|
ResourceID string `json:"resource_id"`
|
|
Permission string `json:"permission"`
|
|
Title string `json:"title,omitempty"`
|
|
Text string `json:"text,omitempty"`
|
|
Turns []SharedTurn `json:"turns,omitempty"`
|
|
AccessURL string `json:"access_url,omitempty"`
|
|
AccessUntil *time.Time `json:"access_until,omitempty"`
|
|
Version int64 `json:"version,omitempty"`
|
|
RevisionNo int64 `json:"revision_no,omitempty"`
|
|
}
|
|
|
|
type resourceBoundary struct{ tenantID, householdID, subjectID, subjectPublicID string }
|
|
|
|
func NewService(pool *pgxpool.Pool, auditRepository *audit.Repository, store mediaSigner, tokenSecret []byte, accessTTL time.Duration) (*Service, error) {
|
|
if pool == nil || auditRepository == nil || store == nil || len(tokenSecret) < 32 || accessTTL <= 0 {
|
|
return nil, errors.New("sharing database, audit, object store, token secret and access TTL are required")
|
|
}
|
|
return &Service{pool: pool, audit: auditRepository, idem: idempotency.NewExecutor(pool), store: store, accessTTL: accessTTL, tokenSecret: append([]byte(nil), tokenSecret...)}, nil
|
|
}
|
|
|
|
func normalizeInput(input CreateInput) (CreateInput, error) {
|
|
input.ResourceType = strings.ToUpper(strings.TrimSpace(input.ResourceType))
|
|
input.Permission = strings.ToUpper(strings.TrimSpace(input.Permission))
|
|
input.Purpose = strings.TrimSpace(input.Purpose)
|
|
valid := (input.ResourceType == "STORY" && (input.Permission == "VIEW_TEXT" || input.Permission == "EDIT")) ||
|
|
(input.ResourceType == "TRANSCRIPT" && input.Permission == "VIEW_TEXT") ||
|
|
(input.ResourceType == "RECORDING" && (input.Permission == "LISTEN" || input.Permission == "DOWNLOAD"))
|
|
if !valid || (!input.LoginRequired && input.Permission == "EDIT") || input.ResourceID == "" || input.GranteeUserID == "" || input.Purpose == "" || len([]rune(input.Purpose)) > 500 || input.ExpiresAt.Before(time.Now().UTC().Add(time.Minute)) {
|
|
return CreateInput{}, ErrInvalid
|
|
}
|
|
return input, nil
|
|
}
|
|
|
|
func (s *Service) Create(ctx context.Context, principal identity.Principal, key, requestID string, raw CreateInput) (Grant, bool, error) {
|
|
input, err := normalizeInput(raw)
|
|
if err != nil {
|
|
return Grant{}, false, err
|
|
}
|
|
boundary, grantorID, grantorPublicID, err := s.resolveGrantor(ctx, s.pool, principal, input.ResourceType, input.ResourceID, false)
|
|
if err != nil {
|
|
return Grant{}, false, err
|
|
}
|
|
canonical, _ := json.Marshal(input)
|
|
grantPublicID := s.deriveGrantID(principal.Issuer + "|" + principal.Subject + "|" + key + "|" + idempotency.Hash(canonical))
|
|
var tokenDigest any
|
|
if !input.LoginRequired {
|
|
tokenDigest = digest(s.deriveAccessToken(grantPublicID))
|
|
}
|
|
result, err := s.idem.Execute(ctx, idempotency.Request{TenantID: &boundary.tenantID, ActorKey: principal.Issuer + "|" + principal.Subject, Scope: "sharing-grant.create", Key: key, RequestHash: idempotency.Hash(canonical), ExpiresAfter: 24 * time.Hour}, func(ctx context.Context, tx pgx.Tx) (idempotency.Result, error) {
|
|
locked, actorID, actorPublicID, err := s.resolveGrantor(ctx, tx, principal, input.ResourceType, input.ResourceID, true)
|
|
if err != nil {
|
|
return idempotency.Result{}, err
|
|
}
|
|
if locked.tenantID != boundary.tenantID || actorID != grantorID {
|
|
return idempotency.Result{}, ErrAccessDenied
|
|
}
|
|
var granteeID string
|
|
if err := tx.QueryRow(ctx, `SELECT id::text FROM user_account WHERE tenant_id=$1::uuid AND public_id=$2::uuid AND status='active'`, locked.tenantID, input.GranteeUserID).Scan(&granteeID); errors.Is(err, pgx.ErrNoRows) {
|
|
return idempotency.Result{}, ErrAccessDenied
|
|
} else if err != nil {
|
|
return idempotency.Result{}, err
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE sharing_grant SET revoked_at=now(),revoked_by=$1::uuid,version=version+1,updated_at=now() WHERE tenant_id=$2::uuid AND resource_type=$3 AND resource_public_id=$4::uuid AND grantee_user_id=$5::uuid AND permission=$6 AND revoked_at IS NULL AND expires_at>now()`, actorID, locked.tenantID, input.ResourceType, input.ResourceID, granteeID, input.Permission); err != nil {
|
|
return idempotency.Result{}, err
|
|
}
|
|
var item Grant
|
|
err = tx.QueryRow(ctx, `INSERT INTO sharing_grant(public_id,tenant_id,household_id,subject_id,resource_type,resource_public_id,grantee_user_id,permission,purpose,login_required,access_token_digest,granted_by,expires_at) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$5,$6::uuid,$7::uuid,$8,$9,$10,$11,$12::uuid,$13) RETURNING public_id::text,$14::text,resource_type,resource_public_id::text,$15::text,permission,purpose,login_required,$16::text,granted_at,expires_at,revoked_at,version`, grantPublicID, locked.tenantID, locked.householdID, locked.subjectID, input.ResourceType, input.ResourceID, granteeID, input.Permission, input.Purpose, input.LoginRequired, tokenDigest, actorID, input.ExpiresAt, locked.subjectPublicID, input.GranteeUserID, actorPublicID).Scan(&item.ID, &item.SubjectID, &item.ResourceType, &item.ResourceID, &item.GranteeUserID, &item.Permission, &item.Purpose, &item.LoginRequired, &item.GrantedBy, &item.GrantedAt, &item.ExpiresAt, &item.RevokedAt, &item.Version)
|
|
if err != nil {
|
|
return idempotency.Result{}, err
|
|
}
|
|
if err := s.audit.Record(ctx, tx, audit.Event{TenantID: &locked.tenantID, ActorType: "user", ActorPublicID: &actorPublicID, Action: "sharing.grant.create", ResourceType: "sharing_grant", ResourceID: &item.ID, RequestID: requestID, Purpose: input.Purpose, Result: "success", Detail: map[string]any{"resource_type": input.ResourceType, "resource_id": input.ResourceID, "grantee_user_id": input.GranteeUserID, "permission": input.Permission, "expires_at": input.ExpiresAt}}); err != nil {
|
|
return idempotency.Result{}, err
|
|
}
|
|
body, _ := json.Marshal(item)
|
|
return idempotency.Result{StatusCode: 201, Body: body, ResourceID: &item.ID}, nil
|
|
})
|
|
if err != nil {
|
|
return Grant{}, false, err
|
|
}
|
|
var item Grant
|
|
if err := json.Unmarshal(result.Body, &item); err != nil {
|
|
return Grant{}, false, err
|
|
}
|
|
if !item.LoginRequired {
|
|
item.AccessToken = s.deriveAccessToken(item.ID)
|
|
}
|
|
_ = grantorPublicID
|
|
return item, !result.Replayed, nil
|
|
}
|
|
|
|
func (s *Service) deriveGrantID(value string) string {
|
|
mac := hmac.New(sha256.New, s.tokenSecret)
|
|
_, _ = mac.Write([]byte("sharing-grant:" + value))
|
|
sum := mac.Sum(nil)
|
|
sum[6] = (sum[6] & 0x0f) | 0x40
|
|
sum[8] = (sum[8] & 0x3f) | 0x80
|
|
return fmt.Sprintf("%x-%x-%x-%x-%x", sum[0:4], sum[4:6], sum[6:8], sum[8:10], sum[10:16])
|
|
}
|
|
|
|
func (s *Service) deriveAccessToken(grantID string) string {
|
|
mac := hmac.New(sha256.New, s.tokenSecret)
|
|
_, _ = mac.Write([]byte("sharing-access:" + grantID))
|
|
return base64.RawURLEncoding.EncodeToString(mac.Sum(nil))
|
|
}
|
|
|
|
func digest(value string) string {
|
|
sum := sha256.Sum256([]byte(value))
|
|
return hex.EncodeToString(sum[:])
|
|
}
|
|
|
|
func (s *Service) List(ctx context.Context, principal identity.Principal, subjectPublicID string) ([]Grant, error) {
|
|
var tenantID, subjectID string
|
|
err := s.pool.QueryRow(ctx, `SELECT sp.tenant_id::text,sp.id::text FROM subject_profile sp JOIN household_member hm ON hm.tenant_id=sp.tenant_id AND hm.household_id=sp.household_id AND hm.status='active' JOIN user_account ua ON ua.id=hm.user_id AND ua.tenant_id=sp.tenant_id AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3 AND ua.status='active' WHERE sp.public_id=$1::uuid AND sp.status='active' AND hm.role IN('owner','admin')`, subjectPublicID, principal.Issuer, principal.Subject).Scan(&tenantID, &subjectID)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, ErrAccessDenied
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := s.pool.Query(ctx, `SELECT sg.public_id::text,$3::text,sg.resource_type,sg.resource_public_id::text,grantee.public_id::text,sg.permission,sg.purpose,sg.login_required,grantor.public_id::text,sg.granted_at,sg.expires_at,sg.revoked_at,sg.version FROM sharing_grant sg JOIN user_account grantee ON grantee.id=sg.grantee_user_id AND grantee.tenant_id=sg.tenant_id JOIN user_account grantor ON grantor.id=sg.granted_by AND grantor.tenant_id=sg.tenant_id WHERE sg.tenant_id=$1::uuid AND sg.subject_id=$2::uuid ORDER BY sg.granted_at DESC LIMIT 200`, tenantID, subjectID, subjectPublicID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := make([]Grant, 0)
|
|
for rows.Next() {
|
|
var item Grant
|
|
if err := rows.Scan(&item.ID, &item.SubjectID, &item.ResourceType, &item.ResourceID, &item.GranteeUserID, &item.Permission, &item.Purpose, &item.LoginRequired, &item.GrantedBy, &item.GrantedAt, &item.ExpiresAt, &item.RevokedAt, &item.Version); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
return items, rows.Err()
|
|
}
|
|
|
|
func (s *Service) Revoke(ctx context.Context, principal identity.Principal, grantPublicID, requestID string) (Grant, error) {
|
|
var item Grant
|
|
tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{})
|
|
if err != nil {
|
|
return item, err
|
|
}
|
|
defer tx.Rollback(context.Background())
|
|
var tenantID, actorID, actorPublicID, role string
|
|
err = tx.QueryRow(ctx, `SELECT sg.tenant_id::text,ua.id::text,ua.public_id::text,hm.role,sg.public_id::text,sp.public_id::text,sg.resource_type,sg.resource_public_id::text,grantee.public_id::text,sg.permission,sg.purpose,sg.login_required,grantor.public_id::text,sg.granted_at,sg.expires_at,sg.revoked_at,sg.version FROM sharing_grant sg JOIN subject_profile sp ON sp.id=sg.subject_id AND sp.tenant_id=sg.tenant_id JOIN household_member hm ON hm.tenant_id=sg.tenant_id AND hm.household_id=sg.household_id AND hm.status='active' JOIN user_account ua ON ua.id=hm.user_id AND ua.tenant_id=sg.tenant_id AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3 AND ua.status='active' JOIN user_account grantee ON grantee.id=sg.grantee_user_id JOIN user_account grantor ON grantor.id=sg.granted_by WHERE sg.public_id=$1::uuid FOR UPDATE OF sg`, grantPublicID, principal.Issuer, principal.Subject).Scan(&tenantID, &actorID, &actorPublicID, &role, &item.ID, &item.SubjectID, &item.ResourceType, &item.ResourceID, &item.GranteeUserID, &item.Permission, &item.Purpose, &item.LoginRequired, &item.GrantedBy, &item.GrantedAt, &item.ExpiresAt, &item.RevokedAt, &item.Version)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return item, ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return item, err
|
|
}
|
|
if role != "owner" && role != "admin" {
|
|
return item, ErrAccessDenied
|
|
}
|
|
if item.RevokedAt == nil {
|
|
err = tx.QueryRow(ctx, `UPDATE sharing_grant SET revoked_at=now(),revoked_by=$2::uuid,version=version+1,updated_at=now() WHERE public_id=$1::uuid RETURNING revoked_at,version`, grantPublicID, actorID).Scan(&item.RevokedAt, &item.Version)
|
|
if err != nil {
|
|
return item, err
|
|
}
|
|
}
|
|
if err = s.audit.Record(ctx, tx, audit.Event{TenantID: &tenantID, ActorType: "user", ActorPublicID: &actorPublicID, Action: "sharing.grant.revoke", ResourceType: "sharing_grant", ResourceID: &item.ID, RequestID: requestID, Result: "success"}); err != nil {
|
|
return item, err
|
|
}
|
|
if err = tx.Commit(ctx); err != nil {
|
|
return item, err
|
|
}
|
|
return item, nil
|
|
}
|
|
|
|
func (s *Service) Access(ctx context.Context, principal identity.Principal, grantPublicID, requestID string) (SharedResource, error) {
|
|
digestBytes := sha256.Sum256([]byte(principal.Issuer + "|" + principal.Subject))
|
|
actorDigest := hex.EncodeToString(digestBytes[:])
|
|
var grantID, tenantID, userID, resourceType, resourceID, permission string
|
|
var allowed, active bool
|
|
err := s.pool.QueryRow(ctx, `SELECT sg.id::text,sg.tenant_id::text,COALESCE(ua.id::text,''),sg.resource_type,sg.resource_public_id::text,sg.permission,COALESCE(ua.id=sg.grantee_user_id,false), (sg.revoked_at IS NULL AND sg.expires_at>now()) FROM sharing_grant sg LEFT JOIN user_account ua ON ua.tenant_id=sg.tenant_id AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3 AND ua.status='active' WHERE sg.public_id=$1::uuid`, grantPublicID, principal.Issuer, principal.Subject).Scan(&grantID, &tenantID, &userID, &resourceType, &resourceID, &permission, &allowed, &active)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return SharedResource{}, ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return SharedResource{}, err
|
|
}
|
|
if !active {
|
|
if logErr := s.logAccess(ctx, tenantID, grantID, userID, actorDigest, resourceType, resourceID, permission, "DENIED", "GRANT_INACTIVE", requestID); logErr != nil {
|
|
return SharedResource{}, logErr
|
|
}
|
|
return SharedResource{}, ErrExpired
|
|
}
|
|
if !allowed {
|
|
if logErr := s.logAccess(ctx, tenantID, grantID, userID, actorDigest, resourceType, resourceID, permission, "DENIED", "RECIPIENT_MISMATCH", requestID); logErr != nil {
|
|
return SharedResource{}, logErr
|
|
}
|
|
return SharedResource{}, ErrAccessDenied
|
|
}
|
|
shared := SharedResource{GrantID: grantPublicID, ResourceType: resourceType, ResourceID: resourceID, Permission: permission}
|
|
switch resourceType {
|
|
case "STORY":
|
|
err = s.pool.QueryRow(ctx, `SELECT sd.title,sr.content,sd.version,sr.revision_no FROM story_draft sd JOIN story_revision sr ON sr.id=sd.current_revision_id WHERE sd.public_id=$1::uuid AND sd.status<>'SUPERSEDED' AND call_story_available(sd.id)`, resourceID).Scan(&shared.Title, &shared.Text, &shared.Version, &shared.RevisionNo)
|
|
case "TRANSCRIPT":
|
|
rows, qerr := s.pool.Query(ctx, `SELECT ct.sequence_no,ct.speaker,ct.start_at,ct.end_at,ct.final_text,ct.interrupted FROM conversation_turn ct JOIN conversation c ON c.id=ct.conversation_id WHERE c.public_id=$1::uuid ORDER BY ct.sequence_no LIMIT 1000`, resourceID)
|
|
if qerr != nil {
|
|
return SharedResource{}, qerr
|
|
}
|
|
defer rows.Close()
|
|
shared.Turns = []SharedTurn{}
|
|
for rows.Next() {
|
|
var turn SharedTurn
|
|
if err := rows.Scan(&turn.SequenceNo, &turn.Speaker, &turn.StartAt, &turn.EndAt, &turn.Text, &turn.Interrupted); err != nil {
|
|
return SharedResource{}, err
|
|
}
|
|
shared.Turns = append(shared.Turns, turn)
|
|
}
|
|
err = rows.Err()
|
|
case "RECORDING":
|
|
var objectRef, contentType, track, conversationPublicID string
|
|
var sequence int
|
|
var consent bool
|
|
err = s.pool.QueryRow(ctx, `SELECT ms.object_ref,ms.content_type,ms.track,ms.sequence_no,c.public_id::text,EXISTS(SELECT 1 FROM consent_record cr WHERE cr.tenant_id=ms.tenant_id AND cr.subject_id=c.subject_id AND cr.consent_type='raw_audio_storage' AND cr.withdrawn_at IS NULL AND (cr.expires_at IS NULL OR cr.expires_at>now())) FROM media_segment ms JOIN conversation c ON c.id=ms.conversation_id WHERE ms.public_id=$1::uuid AND ms.status='VERIFIED'`, resourceID).Scan(&objectRef, &contentType, &track, &sequence, &conversationPublicID, &consent)
|
|
if err == nil && !consent {
|
|
if logErr := s.logAccess(ctx, tenantID, grantID, userID, actorDigest, resourceType, resourceID, permission, "DENIED", "CONSENT_REQUIRED", requestID); logErr != nil {
|
|
return SharedResource{}, logErr
|
|
}
|
|
return SharedResource{}, ErrAccessDenied
|
|
}
|
|
if err == nil {
|
|
extension := ".bin"
|
|
if contentType == "audio/wav" {
|
|
extension = ".wav"
|
|
} else if contentType == "audio/opus" {
|
|
extension = ".opus"
|
|
} else if contentType == "audio/ogg" {
|
|
extension = ".ogg"
|
|
}
|
|
until := time.Now().UTC().Add(s.accessTTL)
|
|
shared.AccessURL, err = s.store.PresignMediaDownload(ctx, objectRef, s.accessTTL, fmt.Sprintf("%s-%s-%08d%s", conversationPublicID, track, sequence, extension), contentType, permission == "LISTEN")
|
|
shared.AccessUntil = &until
|
|
}
|
|
default:
|
|
return SharedResource{}, ErrInvalid
|
|
}
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
if logErr := s.logAccess(ctx, tenantID, grantID, userID, actorDigest, resourceType, resourceID, permission, "DENIED", "RESOURCE_UNAVAILABLE", requestID); logErr != nil {
|
|
return SharedResource{}, logErr
|
|
}
|
|
return SharedResource{}, ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return SharedResource{}, err
|
|
}
|
|
if logErr := s.logAccess(ctx, tenantID, grantID, userID, actorDigest, resourceType, resourceID, permission, "SUCCESS", "", requestID); logErr != nil {
|
|
return SharedResource{}, logErr
|
|
}
|
|
return shared, nil
|
|
}
|
|
|
|
func (s *Service) AccessByToken(ctx context.Context, grantPublicID, token, requestID string) (SharedResource, error) {
|
|
token = strings.TrimSpace(token)
|
|
actorDigest := digest(token)
|
|
var grantID, tenantID, resourceType, resourceID, permission, storedDigest string
|
|
var active, loginRequired bool
|
|
err := s.pool.QueryRow(ctx, `SELECT id::text,tenant_id::text,resource_type,resource_public_id::text,permission,COALESCE(access_token_digest,''),(revoked_at IS NULL AND expires_at>now()),login_required FROM sharing_grant WHERE public_id=$1::uuid`, grantPublicID).Scan(&grantID, &tenantID, &resourceType, &resourceID, &permission, &storedDigest, &active, &loginRequired)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return SharedResource{}, ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return SharedResource{}, err
|
|
}
|
|
validToken := len(token) >= 32 && subtle.ConstantTimeCompare([]byte(actorDigest), []byte(storedDigest)) == 1
|
|
if loginRequired || !validToken {
|
|
if logErr := s.logAccess(ctx, tenantID, grantID, "", actorDigest, resourceType, resourceID, permission, "DENIED", "TOKEN_INVALID", requestID); logErr != nil {
|
|
return SharedResource{}, logErr
|
|
}
|
|
return SharedResource{}, ErrAccessDenied
|
|
}
|
|
if !active {
|
|
if logErr := s.logAccess(ctx, tenantID, grantID, "", actorDigest, resourceType, resourceID, permission, "DENIED", "GRANT_INACTIVE", requestID); logErr != nil {
|
|
return SharedResource{}, logErr
|
|
}
|
|
return SharedResource{}, ErrExpired
|
|
}
|
|
shared, err := s.loadResource(ctx, grantPublicID, resourceType, resourceID, permission)
|
|
if errors.Is(err, pgx.ErrNoRows) || errors.Is(err, ErrNotFound) {
|
|
_ = s.logAccess(ctx, tenantID, grantID, "", actorDigest, resourceType, resourceID, permission, "DENIED", "RESOURCE_UNAVAILABLE", requestID)
|
|
return SharedResource{}, ErrNotFound
|
|
}
|
|
if errors.Is(err, ErrAccessDenied) {
|
|
if logErr := s.logAccess(ctx, tenantID, grantID, "", actorDigest, resourceType, resourceID, permission, "DENIED", "CONSENT_REQUIRED", requestID); logErr != nil {
|
|
return SharedResource{}, logErr
|
|
}
|
|
return SharedResource{}, ErrAccessDenied
|
|
}
|
|
if err != nil {
|
|
return SharedResource{}, err
|
|
}
|
|
if err = s.logAccess(ctx, tenantID, grantID, "", actorDigest, resourceType, resourceID, permission, "SUCCESS", "", requestID); err != nil {
|
|
return SharedResource{}, err
|
|
}
|
|
return shared, nil
|
|
}
|
|
|
|
func (s *Service) loadResource(ctx context.Context, grantPublicID, resourceType, resourceID, permission string) (SharedResource, error) {
|
|
shared := SharedResource{GrantID: grantPublicID, ResourceType: resourceType, ResourceID: resourceID, Permission: permission}
|
|
switch resourceType {
|
|
case "STORY":
|
|
err := s.pool.QueryRow(ctx, `SELECT sd.title,sr.content,sd.version,sr.revision_no FROM story_draft sd JOIN story_revision sr ON sr.id=sd.current_revision_id WHERE sd.public_id=$1::uuid AND sd.status<>'SUPERSEDED' AND call_story_available(sd.id)`, resourceID).Scan(&shared.Title, &shared.Text, &shared.Version, &shared.RevisionNo)
|
|
return shared, err
|
|
case "TRANSCRIPT":
|
|
rows, err := s.pool.Query(ctx, `SELECT ct.sequence_no,ct.speaker,ct.start_at,ct.end_at,ct.final_text,ct.interrupted FROM conversation_turn ct JOIN conversation c ON c.id=ct.conversation_id WHERE c.public_id=$1::uuid ORDER BY ct.sequence_no LIMIT 1000`, resourceID)
|
|
if err != nil {
|
|
return shared, err
|
|
}
|
|
defer rows.Close()
|
|
shared.Turns = []SharedTurn{}
|
|
for rows.Next() {
|
|
var turn SharedTurn
|
|
if err := rows.Scan(&turn.SequenceNo, &turn.Speaker, &turn.StartAt, &turn.EndAt, &turn.Text, &turn.Interrupted); err != nil {
|
|
return shared, err
|
|
}
|
|
shared.Turns = append(shared.Turns, turn)
|
|
}
|
|
return shared, rows.Err()
|
|
case "RECORDING":
|
|
var objectRef, contentType, track, conversationPublicID string
|
|
var sequence int
|
|
var consent bool
|
|
err := s.pool.QueryRow(ctx, `SELECT ms.object_ref,ms.content_type,ms.track,ms.sequence_no,c.public_id::text,EXISTS(SELECT 1 FROM consent_record cr WHERE cr.tenant_id=ms.tenant_id AND cr.subject_id=c.subject_id AND cr.consent_type='raw_audio_storage' AND cr.withdrawn_at IS NULL AND (cr.expires_at IS NULL OR cr.expires_at>now())) FROM media_segment ms JOIN conversation c ON c.id=ms.conversation_id WHERE ms.public_id=$1::uuid AND ms.status='VERIFIED'`, resourceID).Scan(&objectRef, &contentType, &track, &sequence, &conversationPublicID, &consent)
|
|
if err != nil {
|
|
return shared, err
|
|
}
|
|
if !consent {
|
|
return shared, ErrAccessDenied
|
|
}
|
|
extension := ".bin"
|
|
if contentType == "audio/wav" {
|
|
extension = ".wav"
|
|
} else if contentType == "audio/opus" {
|
|
extension = ".opus"
|
|
} else if contentType == "audio/ogg" {
|
|
extension = ".ogg"
|
|
}
|
|
until := time.Now().UTC().Add(s.accessTTL)
|
|
shared.AccessURL, err = s.store.PresignMediaDownload(ctx, objectRef, s.accessTTL, fmt.Sprintf("%s-%s-%08d%s", conversationPublicID, track, sequence, extension), contentType, permission == "LISTEN")
|
|
shared.AccessUntil = &until
|
|
return shared, err
|
|
default:
|
|
return shared, ErrInvalid
|
|
}
|
|
}
|
|
|
|
func (s *Service) EditStory(ctx context.Context, principal identity.Principal, grantPublicID, content, reason, requestID string, expectedVersion int64) (SharedResource, error) {
|
|
content = strings.TrimSpace(content)
|
|
reason = strings.TrimSpace(reason)
|
|
if content == "" || len([]rune(content)) > 50000 || reason == "" || len([]rune(reason)) > 500 || expectedVersion < 1 {
|
|
return SharedResource{}, ErrInvalid
|
|
}
|
|
tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{})
|
|
if err != nil {
|
|
return SharedResource{}, err
|
|
}
|
|
defer tx.Rollback(context.Background())
|
|
var callSubject string
|
|
err = tx.QueryRow(ctx, `SELECT sd.subject_id::text FROM sharing_grant sg
|
|
JOIN user_account ua ON ua.id=sg.grantee_user_id AND ua.tenant_id=sg.tenant_id AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3 AND ua.status='active'
|
|
JOIN story_draft sd ON sd.public_id=sg.resource_public_id AND sd.tenant_id=sg.tenant_id
|
|
WHERE sg.public_id=$1::uuid AND sg.resource_type='STORY' AND sg.permission='EDIT' AND sg.revoked_at IS NULL AND sg.expires_at>now() AND sd.call_session_id IS NOT NULL`, grantPublicID, principal.Issuer, principal.Subject).Scan(&callSubject)
|
|
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
|
|
return SharedResource{}, err
|
|
}
|
|
if callSubject != "" {
|
|
var epoch int64
|
|
if err = tx.QueryRow(ctx, `SELECT privacy_epoch FROM memory_user_state WHERE subject_id=$1::uuid FOR UPDATE`, callSubject).Scan(&epoch); err != nil {
|
|
return SharedResource{}, err
|
|
}
|
|
}
|
|
var grantID, tenantID, userID, userPublicID, resourceID, permission, storyID, currentRevisionID, title string
|
|
var active bool
|
|
var version, revisionNo int64
|
|
err = tx.QueryRow(ctx, `SELECT sg.id::text,sg.tenant_id::text,ua.id::text,ua.public_id::text,sg.resource_public_id::text,sg.permission,(sg.revoked_at IS NULL AND sg.expires_at>now()),sd.id::text,sd.current_revision_id::text,sd.title,sd.version,sr.revision_no FROM sharing_grant sg JOIN user_account ua ON ua.id=sg.grantee_user_id AND ua.tenant_id=sg.tenant_id AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3 AND ua.status='active' JOIN story_draft sd ON sd.public_id=sg.resource_public_id AND sd.tenant_id=sg.tenant_id JOIN story_revision sr ON sr.id=sd.current_revision_id WHERE sg.public_id=$1::uuid AND sg.resource_type='STORY' AND call_story_available(sd.id) FOR UPDATE OF sg,sd`, grantPublicID, principal.Issuer, principal.Subject).Scan(&grantID, &tenantID, &userID, &userPublicID, &resourceID, &permission, &active, &storyID, ¤tRevisionID, &title, &version, &revisionNo)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return SharedResource{}, ErrAccessDenied
|
|
}
|
|
if err != nil {
|
|
return SharedResource{}, err
|
|
}
|
|
if !active {
|
|
return SharedResource{}, ErrExpired
|
|
}
|
|
if permission != "EDIT" {
|
|
return SharedResource{}, ErrAccessDenied
|
|
}
|
|
if version != expectedVersion {
|
|
return SharedResource{}, ErrVersionConflict
|
|
}
|
|
var available bool
|
|
if err = tx.QueryRow(ctx, `SELECT call_story_available(id) AND status NOT IN ('REJECTED','SUPERSEDED') FROM story_draft WHERE id=$1::uuid`, storyID).Scan(&available); err != nil {
|
|
return SharedResource{}, err
|
|
}
|
|
if !available {
|
|
return SharedResource{}, ErrAccessDenied
|
|
}
|
|
var newRevisionID string
|
|
err = tx.QueryRow(ctx, `INSERT INTO story_revision(story_id,revision_no,content,author_type,author_id,edit_reason,base_revision_id) VALUES($1::uuid,$2,$3,'user',$4::uuid,$5,$6::uuid) RETURNING id::text`, storyID, revisionNo+1, content, userID, reason, currentRevisionID).Scan(&newRevisionID)
|
|
if err != nil {
|
|
return SharedResource{}, err
|
|
}
|
|
if _, err = tx.Exec(ctx, `INSERT INTO story_source(story_revision_id,conversation_turn_id,call_session_id,call_utterance_id,call_source_revision,import_turn_id,source_start_offset,source_end_offset,source_role) SELECT $1::uuid,conversation_turn_id,call_session_id,call_utterance_id,call_source_revision,import_turn_id,source_start_offset,source_end_offset,source_role FROM story_source WHERE story_revision_id=$2::uuid`, newRevisionID, currentRevisionID); err != nil {
|
|
return SharedResource{}, err
|
|
}
|
|
if err = tx.QueryRow(ctx, `UPDATE story_draft SET current_revision_id=$2::uuid,status='IN_REVIEW',reviewed_by=$3::uuid,reviewed_at=now(),review_reason=$4,version=version+1,updated_at=now() WHERE id=$1::uuid RETURNING version`, storyID, newRevisionID, userID, reason).Scan(&version); err != nil {
|
|
return SharedResource{}, err
|
|
}
|
|
digestBytes := sha256.Sum256([]byte(principal.Issuer + "|" + principal.Subject))
|
|
actorDigest := hex.EncodeToString(digestBytes[:])
|
|
if _, err = tx.Exec(ctx, `INSERT INTO sharing_access_log(tenant_id,sharing_grant_id,actor_user_id,actor_identifier_digest,resource_type,resource_public_id,permission,result,request_id) VALUES($1::uuid,$2::uuid,$3::uuid,$4,'STORY',$5::uuid,'EDIT','SUCCESS',$6)`, tenantID, grantID, userID, actorDigest, resourceID, requestID); err != nil {
|
|
return SharedResource{}, err
|
|
}
|
|
if err = s.audit.Record(ctx, tx, audit.Event{TenantID: &tenantID, ActorType: "user", ActorPublicID: &userPublicID, Action: "sharing.story.edit", ResourceType: "story", ResourceID: &resourceID, RequestID: requestID, Purpose: reason, Result: "success", Detail: map[string]any{"sharing_grant_id": grantPublicID, "revision_no": revisionNo + 1}}); err != nil {
|
|
return SharedResource{}, err
|
|
}
|
|
if err = tx.Commit(ctx); err != nil {
|
|
return SharedResource{}, err
|
|
}
|
|
return SharedResource{GrantID: grantPublicID, ResourceType: "STORY", ResourceID: resourceID, Permission: "EDIT", Title: title, Text: content, Version: version, RevisionNo: revisionNo + 1}, nil
|
|
}
|
|
|
|
func (s *Service) logAccess(ctx context.Context, tenantID, grantID, userID, actorDigest, resourceType, resourceID, permission, result, reason, requestID string) error {
|
|
var actor any
|
|
if userID != "" {
|
|
actor = userID
|
|
}
|
|
_, err := s.pool.Exec(ctx, `INSERT INTO sharing_access_log(tenant_id,sharing_grant_id,actor_user_id,actor_identifier_digest,resource_type,resource_public_id,permission,result,reason_code,request_id) VALUES($1::uuid,$2::uuid,$3::uuid,$4,$5,$6::uuid,$7,$8,NULLIF($9,''),$10)`, tenantID, grantID, actor, actorDigest, resourceType, resourceID, permission, result, reason, requestID)
|
|
return err
|
|
}
|
|
|
|
type queryer interface {
|
|
QueryRow(context.Context, string, ...any) pgx.Row
|
|
}
|
|
|
|
func (s *Service) resolveGrantor(ctx context.Context, q queryer, principal identity.Principal, resourceType, resourcePublicID string, lock bool) (resourceBoundary, string, string, error) {
|
|
resourceTable := ""
|
|
joins := ""
|
|
subjectExpr := ""
|
|
householdExpr := ""
|
|
tenantExpr := ""
|
|
switch resourceType {
|
|
case "STORY":
|
|
resourceTable = "story_draft resource"
|
|
joins = "JOIN subject_profile sp ON sp.id=resource.subject_id AND sp.tenant_id=resource.tenant_id"
|
|
subjectExpr = "resource.subject_id"
|
|
householdExpr = "sp.household_id"
|
|
tenantExpr = "resource.tenant_id"
|
|
case "TRANSCRIPT":
|
|
resourceTable = "conversation resource"
|
|
joins = "JOIN subject_profile sp ON sp.id=resource.subject_id AND sp.tenant_id=resource.tenant_id"
|
|
subjectExpr = "resource.subject_id"
|
|
householdExpr = "resource.household_id"
|
|
tenantExpr = "resource.tenant_id"
|
|
case "RECORDING":
|
|
resourceTable = "media_segment resource"
|
|
joins = "JOIN conversation c ON c.id=resource.conversation_id AND c.tenant_id=resource.tenant_id JOIN subject_profile sp ON sp.id=c.subject_id AND sp.tenant_id=c.tenant_id"
|
|
subjectExpr = "c.subject_id"
|
|
householdExpr = "c.household_id"
|
|
tenantExpr = "resource.tenant_id"
|
|
default:
|
|
return resourceBoundary{}, "", "", ErrInvalid
|
|
}
|
|
lockClause := ""
|
|
if resourceType == "STORY" {
|
|
lockClause = " AND call_story_available(resource.id)"
|
|
}
|
|
if lock {
|
|
lockClause += " FOR SHARE OF resource"
|
|
}
|
|
query := fmt.Sprintf(`SELECT %s::text,%s::text,%s::text,sp.public_id::text,ua.id::text,ua.public_id::text,hm.role FROM %s %s JOIN household_member hm ON hm.tenant_id=%s AND hm.household_id=%s AND hm.status='active' JOIN user_account ua ON ua.id=hm.user_id AND ua.tenant_id=%s AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3 AND ua.status='active' WHERE resource.public_id=$1::uuid%s`, tenantExpr, householdExpr, subjectExpr, resourceTable, joins, tenantExpr, householdExpr, tenantExpr, lockClause)
|
|
var b resourceBoundary
|
|
var actorID, actorPublicID, role string
|
|
err := q.QueryRow(ctx, query, resourcePublicID, principal.Issuer, principal.Subject).Scan(&b.tenantID, &b.householdID, &b.subjectID, &b.subjectPublicID, &actorID, &actorPublicID, &role)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return b, "", "", ErrNotFound
|
|
}
|
|
if err != nil {
|
|
return b, "", "", err
|
|
}
|
|
if role != "owner" && role != "admin" {
|
|
return b, "", "", ErrAccessDenied
|
|
}
|
|
return b, actorID, actorPublicID, nil
|
|
}
|