dr.ptAPRENDER. PRATICAR. EVOLUIR.
← CCSP: segurança cloud, dados e operação
11 / 18 · 55 MIN

Portabilidade: dados, relações e acessos

Executa exportação e restauro, detetando perdas de significado e regressões de acesso.

Define o que deve sobreviver à saída

Num serviço fictício de fundos, o contrato permite exportar posições em CSV. Antes de concluir que existe portabilidade, define o que o destino deve reconstruir: identificadores, tenant, relações com fundos, moeda, unidades monetárias, estados nulos e permissões de leitura. A aplicação pode conseguir comunicar com uma API sem conseguir substituir o serviço que a fornece. A decisão de arquitetura deve identificar dependências de identidade, chaves, agendamentos, logs e formatos proprietários. Regista também o que fica fora do primeiro ensaio. O nosso laboratório cobre três tabelas e um caminho de autorização local; não representa uma migração de fornecedor nem uma conversão entre motores de base de dados.

Usa um contrato de dados explícito

O script exporta funds, positions e grants numa transação de leitura, com versão de esquema, cabeçalhos e hashes dos ficheiros. Os montantes são inteiros em unidades menores e conservam a moeda; não somes EUR e JPY como se fossem a mesma unidade. O valor sintético 9007199254740993 foi escolhido para revelar perda de precisão ao passar por float, não como saldo realista de um cliente. Um indicador separado preserva a diferença entre nota nula e texto vazio. O módulo csv trata vírgulas e quebras de linha segundo o formato definido. Na migração real, acrescenta ainda contratos para datas, fusos horários, escalas decimais e evolução de esquema que o exercício não implementa.

Confirma relações e importação atómica

O destino ativa foreign_keys antes de começar a transação e confirma o estado. As relações usam identificador e tenant: um fundo f1 de A não substitui f1 de B. A importação carrega primeiro fundos, depois posições e finalmente permissões. Se uma posição referenciar um fundo inexistente, a operação falha e o script executa ROLLBACK sobre toda a importação. O ensaio verifica que não ficaram linhas parciais, incluindo as que tinham sido inseridas antes do erro. Não presumas que a falha de uma instrução SQL reverte automaticamente todas as instruções anteriores. Fora deste exemplo, valida também as regras do motor escolhido, os índices e o comportamento de grandes volumes.

Reconcilia significado e acesso

Executa python3 run.py com o código completo abaixo. O import correto conserva linhas, totais por tenant e moeda, relações e a matriz de cinco pedidos. Três variantes mantêm contagens, mas introduzem erros diferentes: um montante perde uma unidade menor, uma nota nula passa a vazia e a permissão de Alice muda de p17 para p18. Os hashes são recalculados de propósito nessas variantes, demonstrando que integridade de transporte não prova correção da transformação. A matriz inclui acessos permitidos e recusados; verificar apenas que Alice consegue entrar não detetaria toda a regressão. As identidades são entradas confiáveis fornecidas ao exercício, sem autenticação implementada.

Transforma resultados em critérios de aceitação

As duas execuções registadas passaram 28 verificações cada em Python 3.13.1 e SQLite 3.53.4. Incluem rejeição de ficheiro alterado, tabela em falta, esquema desconhecido e relação órfã. O resultado serve como evidência dos casos executados; não demonstra desempenho, políticas IAM equivalentes, aplicação em produção ou eliminação no fornecedor anterior. Para o projeto, define quem valida reconciliação financeira, quem aprova permissões e quem opera o destino. Um hash que acompanha o próprio ficheiro não autentica a origem do manifesto. Mantém a referência de comparação sob controlo apropriado e regista versão e âmbito. Relaciona o ensaio com gestão de chaves, recuperação e avaliação de fornecedores.

#!/usr/bin/env python3
"""Original dr.pt portability exercise: synthetic SQLite -> CSV -> SQLite.
No network, cloud APIs, credentials or external packages. Not a migration tool.
"""
import copy
import csv
import datetime
import hashlib
import io
import json
from pathlib import Path
import platform
import sqlite3

checks = []
def check(name, inputs, actual, expected):
    if actual != expected:
        raise AssertionError(f"{name}: {actual!r} != {expected!r}")
    checks.append(dict(name=name, inputs=inputs, actual=actual, expected=expected))

SCHEMA = """
CREATE TABLE funds(id TEXT NOT NULL, tenant TEXT NOT NULL, name TEXT NOT NULL,
 PRIMARY KEY(id,tenant));
CREATE TABLE positions(id TEXT NOT NULL, tenant TEXT NOT NULL, fund_id TEXT NOT NULL,
 currency TEXT NOT NULL, amount_minor INTEGER NOT NULL, note TEXT,
 PRIMARY KEY(id,tenant), FOREIGN KEY(fund_id,tenant) REFERENCES funds(id,tenant));
CREATE TABLE grants(subject TEXT NOT NULL, tenant TEXT NOT NULL, position_id TEXT NOT NULL,
 action TEXT NOT NULL CHECK(action IN ('read','write')),
 PRIMARY KEY(subject,tenant,position_id,action),
 FOREIGN KEY(position_id,tenant) REFERENCES positions(id,tenant));
"""
COLUMNS = {
 'funds': ['id','tenant','name'],
 'positions': ['id','tenant','fund_id','currency','amount_minor','note','note_is_null'],
 'grants': ['subject','tenant','position_id','action']
}
ORDER = {'funds':'tenant,id', 'positions':'tenant,id', 'grants':'tenant,subject,position_id,action'}
def database():
    db = sqlite3.connect(':memory:', isolation_level=None)
    db.execute('PRAGMA foreign_keys=ON')
    if db.execute('PRAGMA foreign_keys').fetchone()[0] != 1:
        raise RuntimeError('Foreign keys unavailable')
    db.executescript(SCHEMA)
    return db

def snapshot(db):
    return {t:db.execute(f'SELECT * FROM {t} ORDER BY {ORDER[t]}').fetchall() for t in COLUMNS}

def digest(data):
    return hashlib.sha256(data).hexdigest()

def csv_bytes(table, rows):
    out = io.StringIO(newline='')
    writer = csv.writer(out, lineterminator='\n')
    writer.writerow(COLUMNS[table])
    for row in rows:
        values = list(row)
        if table == 'positions':
            values.append('1' if values[5] is None else '0')
        writer.writerow(values)
    return out.getvalue().encode('utf-8')

def export(db):
    db.execute('BEGIN')
    try:
        state = snapshot(db)
        bundle = {'schemaVersion':1, 'files':{}, 'manifest':{}}
        for t, rows in state.items():
            data = csv_bytes(t, rows)
            bundle['files'][t] = data
            bundle['manifest'][t] = {'rows':len(rows),'sha256':digest(data)}
        db.execute('COMMIT')
        return bundle
    except Exception:
        db.execute('ROLLBACK')
        raise

def load(bundle, db):
    if bundle['schemaVersion'] != 1 or set(bundle['files']) != set(COLUMNS):
        raise ValueError('Unsupported schema or missing table')
    # Hashes check transport consistency, not authenticity of the manifest.
    for t, data in bundle['files'].items():
        if digest(data) != bundle['manifest'][t]['sha256']:
            raise ValueError('Transport digest mismatch')
    db.execute('BEGIN')
    try:
        for t in COLUMNS: # parents, children, then grants
            reader = csv.reader(io.StringIO(bundle['files'][t].decode('utf-8'), newline=''))
            if next(reader) != COLUMNS[t]:
                raise ValueError('Column mismatch')
            rows = list(reader)
            if len(rows) != bundle['manifest'][t]['rows']:
                raise ValueError('Row count mismatch')
            for row in rows:
                if len(row) != len(COLUMNS[t]):
                    raise ValueError('Column count mismatch')
                if t == 'positions':
                    if row[6] not in ('0','1') or (row[6]=='1' and row[5]!=''):
                        raise ValueError('Null flag mismatch')
                    amount = int(row[4])
                    if str(amount) != row[4]:
                        raise ValueError('Noncanonical integer')
                    row = row[:4]+[amount, None if row[6]=='1' else row[5]]
                db.execute(f"INSERT INTO {t} VALUES ({','.join('?' for _ in row)})", row)
        if db.execute('PRAGMA foreign_key_check').fetchall():
            raise ValueError('Foreign key check failed')
        db.execute('COMMIT')
    except Exception:
        db.execute('ROLLBACK')
        raise

def allowed(db, subject, trusted_tenant, position_id, object_tenant, action):
    # Identity and tenant are supplied as trusted inputs, not authenticated here.
    if trusted_tenant != object_tenant:
        return False
    return db.execute('''SELECT 1 FROM positions p JOIN grants g
      ON g.position_id=p.id AND g.tenant=p.tenant
      WHERE p.id=? AND p.tenant=? AND g.subject=? AND g.action=?''',
      (position_id, object_tenant, subject, action)).fetchone() is not None

REQUESTS = [
 ('alice','A','p17','A','read'),
 ('alice','A','p18','A','read'),
 ('alice','A','p17','B','read'),
 ('alice','A','p17','A','write'),
 ('bob','B','p17','B','read')
]
def totals(db):
    return db.execute('''SELECT tenant,currency,SUM(amount_minor) FROM positions
                        GROUP BY tenant,currency ORDER BY tenant,currency''').fetchall()

def assess(src, dst):
    a,b = snapshot(src),snapshot(dst)
    return {
      'rowCountsMatch':{t:len(v) for t,v in a.items()}=={t:len(v) for t,v in b.items()},
      'canonicalRowsMatch':a==b,
      'amountsByTenantCurrencyMatch':totals(src)==totals(dst),
      'authorizationMatrixMatches':[allowed(src,*q) for q in REQUESTS]==[allowed(dst,*q) for q in REQUESTS],
      'foreignKeysValid':not dst.execute('PRAGMA foreign_key_check').fetchall()
    }

def mutate_csv(bundle, table, transform, refresh_digest=True):
    b = copy.deepcopy(bundle)
    rows = list(csv.reader(io.StringIO(b['files'][table].decode('utf-8'), newline='')))
    transform(rows)
    out = io.StringIO(newline=''); csv.writer(out,lineterminator='\n').writerows(rows)
    b['files'][table] = out.getvalue().encode('utf-8')
    if refresh_digest:
        b['manifest'][table] = {'rows':len(rows)-1,'sha256':digest(b['files'][table])}
    return b

src = database()
src.executemany('INSERT INTO funds VALUES(?,?,?)',[('f1','A','Fundo A, Lisboa'),('f1','B','Fund B')])
src.executemany('INSERT INTO positions VALUES(?,?,?,?,?,?)',[
 ('p17','A','f1','EUR',9007199254740993,None),
 ('p18','A','f1','EUR',700000001,''),
 ('p17','B','f1','JPY',2345,'Liquidação, Lisboa\nLine 2')
])
src.executemany('INSERT INTO grants VALUES(?,?,?,?)',[('alice','A','p17','read'),('bob','B','p17','read')])
bundle = export(src)
dst = database(); load(bundle,dst)
baseline = assess(src,dst)
check('complete export restores all defined invariants',{'schemaVersion':1,'tables':list(COLUMNS)},baseline,dict.fromkeys(baseline,True))
check('explicit foreign keys enabled',{'database':'restored'},dst.execute('PRAGMA foreign_keys').fetchone()[0],1)
check('integer above binary64 exact range preserved',{'column':'amount_minor'},str(dst.execute("SELECT amount_minor FROM positions WHERE tenant='A' AND id='p17'").fetchone()[0]),'9007199254740993')
check('NULL and empty text remain distinct',{'tenant':'A'},[x[0] for x in dst.execute("SELECT note FROM positions WHERE tenant='A' ORDER BY id")],[None,''])
check('comma newline and Unicode survive CSV',{'tenant':'B'},dst.execute("SELECT note FROM positions WHERE tenant='B'").fetchone()[0],'Liquidação, Lisboa\nLine 2')
check('authorization matrix preserved',{'requests':[list(q) for q in REQUESTS]},[allowed(dst,*q) for q in REQUESTS],[True,False,False,False,True])

# Valid transport hashes do not prove that a conversion preserved meaning.
rounded = mutate_csv(bundle,'positions',lambda rows:rows[1].__setitem__(4,str(int(float(rows[1][4])))))
lossy = database(); load(rounded,lossy); verdict = assess(src,lossy)
check('rounded import retains row counts',{'mutation':'binary64 conversion'},verdict['rowCountsMatch'],True)
check('rounded import changes one minor unit',{'mutation':'binary64 conversion'},str(lossy.execute("SELECT amount_minor FROM positions WHERE tenant='A' AND id='p17'").fetchone()[0]),'9007199254740992')
check('amount reconciliation detects rounding',{'mutation':'binary64 conversion'},verdict['amountsByTenantCurrencyMatch'],False)
check('full row comparison detects rounding',{'mutation':'binary64 conversion'},verdict['canonicalRowsMatch'],False)

collapsed = mutate_csv(bundle,'positions',lambda rows:rows[1].__setitem__(6,'0'))
null_loss = database(); load(collapsed,null_loss); verdict = assess(src,null_loss)
check('NULL collapse retains counts and totals',{'mutation':'NULL becomes empty'},[verdict['rowCountsMatch'],verdict['amountsByTenantCurrencyMatch']],[True,True])
check('full row comparison detects NULL collapse',{'mutation':'NULL becomes empty'},verdict['canonicalRowsMatch'],False)

grant_change = mutate_csv(bundle,'grants',lambda rows:rows[1].__setitem__(2,'p18'))
wrong_grants = database(); load(grant_change,wrong_grants); verdict = assess(src,wrong_grants)
check('changed grants retain counts and amounts',{'mutation':'alice grant moved p17 to p18'},[verdict['rowCountsMatch'],verdict['amountsByTenantCurrencyMatch']],[True,True])
check('negative access test detects new permission',{'request':list(REQUESTS[1])},allowed(wrong_grants,*REQUESTS[1]),True)
check('authorization matrix detects grant regression',{'mutation':'alice grant moved p17 to p18'},verdict['authorizationMatrixMatches'],False)

for name,bad in [
 ('orphan relationship',mutate_csv(bundle,'positions',lambda rows:rows[2].__setitem__(2,'missing-fund'))),
 ('transport corruption',mutate_csv(bundle,'positions',lambda rows:rows[2].__setitem__(4,'1'),False)),
 ('missing grants file',{**bundle,'files':{k:v for k,v in bundle['files'].items() if k!='grants'}}),
 ('unsupported schema',{**bundle,'schemaVersion':2})
]:
    candidate = database()
    try:
        load(bad,candidate); rejected=False
    except (ValueError,sqlite3.IntegrityError):
        rejected=True
    check(name+' rejected',{'fault':name},rejected,True)
    check(name+' leaves no partial import',{'fault':name},sum(len(v) for v in snapshot(candidate).values()),0)
    candidate.close()

# Demonstrate why a source snapshot is stale after new destination writes.
dst.execute("UPDATE positions SET amount_minor=amount_minor+7 WHERE tenant='B' AND id='p17'")
check('destination write diverges from source snapshot',{'newDestinationMinorUnits':7},assess(src,dst)['amountsByTenantCurrencyMatch'],False)
check('source snapshot omits post-cutover write',{'tenant':'B'},src.execute("SELECT amount_minor FROM positions WHERE tenant='B'").fetchone()[0],2345)
check('destination contains post-cutover write',{'tenant':'B'},dst.execute("SELECT amount_minor FROM positions WHERE tenant='B'").fetchone()[0],2352)

# This is an original planning calculation, not observed elapsed recovery time.
window, decision, restore, validation = 120,80,25,20
check('late rollback decision exceeds window',{'windowMinutes':window,'decisionMinute':decision,'restoreMinutes':restore,'validationMinutes':validation},decision+restore+validation>window,True)
check('latest calculated rollback decision',{'windowMinutes':window,'restoreMinutes':restore,'validationMinutes':validation},window-restore-validation,75)

for db in (src,dst,lossy,null_loss,wrong_grants):
    db.close()
print(json.dumps(dict(checkedAt=datetime.datetime.now(datetime.timezone.utc).isoformat(),python=platform.python_version(),sqlite=sqlite3.sqlite_version,scriptSha256=digest(Path(__file__).read_bytes()),network=False,groups=['SQLite/CSV semantic restoration','Authorization preservation','Failed-import atomicity','Post-cutover divergence and planning'],checks=checks,limitations=[
 'Actual in-memory SQLite and CSV operations; no cloud-provider migration, database-engine conversion, production application, IAM or network execution.',
 'Synthetic data only. Authorization uses supplied trusted identities and one local SQL path; authentication and provider policy translation are not implemented.',
 'Transport hashes are not signed and do not authenticate the manifest. Mutated bundles deliberately recompute hashes to demonstrate semantic failures.',
 'No concurrent writes, load test, measured RTO/RPO, timed cutover or automatic reverse replication. Rollback timing is an explicit planning calculation.',
 'No physical deletion, legal compliance attestation, contractual enforcement or independent audit demonstrated.'
]),ensure_ascii=False,indent=2))
NA PRÁTICA

Uma contagem correta de posições pode coexistir com montantes ou permissões errados.

Armadilhas comuns

Confundir exportação com portabilidade; hashes com equivalência funcional; infraestrutura antiga com rollback seguro; estimativas com tempos medidos.

Tópicos relacionados: Gestão de chaves e restauro · Risco, fornecedores e continuidade

Leva esta ideia contigo

Executa exportação e restauro, detetando perdas de significado e regressões de acesso.

Criar conta
AnteriorPróxima aula

Referência: SQLite foreign key support · CCSP examination outline effective 2026-08-01; January2026 V2 PDF

CCSP® é uma marca registada de ISC2, Inc. A dr.pt é uma plataforma de preparação independente e não está afiliada, associada, patrocinada, autorizada nem aprovada por ISC2. Os conteúdos e as perguntas são originais, não são perguntas oficiais de exame, e concluir os nossos testes não atribui nem garante qualquer certificação. Os nomes são usados apenas para identificar o tema. Todas as outras marcas pertencem aos respetivos titulares.