1package vnext23import (4 "archive/zip"5 "bytes"6 "crypto/sha256"7 "encoding/hex"8 "encoding/json"9 "fmt"10 "io"11 "math"12 "os"13 "slices"14 "strings"15 "time"16)1718const (19 manifestName = "atlas.json"20 containerFormat = "atlas-schema-bundle"21 minimumFraming = uint16(1)22 containerFraming = uint16(2)23 maxManifestSize = 32 << 2024)2526type schemaReference struct {27 Name string `json:"name"`28 Hash string `json:"hash"`29}3031type tableReference struct {32 Name string `json:"name"`33 RootType ID `json:"rootType"`34 Hash string `json:"hash"`35 Length int64 `json:"length"`36}3738type blobReference struct {39 Name string `json:"name"`40 Hash string `json:"hash"`41 Length int64 `json:"length"`42}4344type bootstrap struct {45 Format string `json:"format"`46 Framing uint16 `json:"framing"`47 Volume string `json:"volume"`48 Release Release `json:"release"`49 Schema schemaReference `json:"schema"`50 Tables []tableReference `json:"tables"`51 Blobs []blobReference `json:"blobs,omitempty"`52}5354type preparedEntry struct {55 name string56 data []byte57 sourcePath string58 method uint1659}6061// Limits bounds the native container before any referenced entry is62// materialized. Zero fields take the stable application defaults; negative63// fields are invalid. Manifests may describe less, never silently more.64type Limits struct {65 MaxEntries int66 MaxTables int67 MaxBlobs int68 MaxTableBytes int6469 MaxBlobBytes int6470 MaxTotalBytes int6471}7273func DefaultLimits() Limits {74 return Limits{75 MaxEntries: 100_000, MaxTables: 65_536, MaxBlobs: 90_000,76 MaxTableBytes: 512 << 20, MaxBlobBytes: 512 << 20, MaxTotalBytes: 1 << 40,77 }78}7980func (limits Limits) resolved() (Limits, error) {81 defaults := DefaultLimits()82 values := []*int{&limits.MaxEntries, &limits.MaxTables, &limits.MaxBlobs}83 defaultInts := []int{defaults.MaxEntries, defaults.MaxTables, defaults.MaxBlobs}84 for index, value := range values {85 if *value < 0 {86 return Limits{}, fmt.Errorf("native container limit is negative")87 }88 if *value == 0 {89 *value = defaultInts[index]90 }91 }92 byteValues := []*int64{&limits.MaxTableBytes, &limits.MaxBlobBytes, &limits.MaxTotalBytes}93 defaultBytes := []int64{defaults.MaxTableBytes, defaults.MaxBlobBytes, defaults.MaxTotalBytes}94 for index, value := range byteValues {95 if *value < 0 {96 return Limits{}, fmt.Errorf("native container byte limit is negative")97 }98 if *value == 0 {99 *value = defaultBytes[index]100 }101 }102 return limits, nil103}104105// Write produces one deterministic ZIP container with a small JSON bootstrap,106// an embedded canonical schema, stored typed blocks, and stored opaque blobs.107func Write(out io.Writer, bundle Bundle) error {108 if err := ValidSlug(bundle.VolumeID); err != nil {109 return fmt.Errorf("volume identity: %w", err)110 }111 if err := bundle.Release.validate(false); err != nil {112 return fmt.Errorf("release: %w", err)113 }114 // The stamp is always derived here; callers cannot smuggle a stale identity115 // across a changed schema or blob set.116 bundle.Release.Stamp = ""117 packedBlobs, err := packRasterBlobs(bundle.Blobs)118 if err != nil {119 return err120 }121 defer cleanupTemporaryBlobs(packedBlobs)122 bundle.Blobs = packedBlobs123 schemaData, err := bundle.Schema.Canonical()124 if err != nil {125 return fmt.Errorf("canonicalize schema: %w", err)126 }127 schemaHash := sha256.Sum256(schemaData)128 schemaName := "schema/" + hex.EncodeToString(schemaHash[:]) + ".json"129 manifest, entries, err := prepareBundle(bundle, schemaHash, schemaName, schemaData)130 if err != nil {131 return err132 }133 stampInput, err := json.Marshal(manifest)134 if err != nil {135 return fmt.Errorf("marshal stamp input: %w", err)136 }137 stamp := sha256.Sum256(stampInput)138 manifest.Release.Stamp = hex.EncodeToString(stamp[:])139 manifestData, err := json.Marshal(manifest)140 if err != nil {141 return fmt.Errorf("marshal bootstrap: %w", err)142 }143 entries = append(entries, preparedEntry{name: manifestName, data: manifestData, method: zip.Deflate})144 slices.SortFunc(entries, func(left, right preparedEntry) int {145 if left.name == manifestName {146 return -1147 }148 if right.name == manifestName {149 return 1150 }151 return strings.Compare(left.name, right.name)152 })153 return writeArchive(out, entries)154}155156func prepareBundle(bundle Bundle, schemaHash [32]byte, schemaName string, schemaData []byte) (bootstrap, []preparedEntry, error) {157 manifest := bootstrap{158 Format: containerFormat, Framing: containerFraming, Volume: bundle.VolumeID,159 Release: bundle.Release,160 Schema: schemaReference{Name: schemaName, Hash: hex.EncodeToString(schemaHash[:])},161 }162 entries := []preparedEntry{{name: schemaName, data: schemaData, method: zip.Deflate}}163 for _, named := range bundle.Tables {164 data, err := EncodeBlock(schemaHash, named.Table)165 if err != nil {166 return bootstrap{}, nil, fmt.Errorf("encode %s: %w", named.Name, err)167 }168 digest := sha256.Sum256(data)169 manifest.Tables = append(manifest.Tables, tableReference{170 Name: named.Name, RootType: named.Table.TypeID,171 Hash: hex.EncodeToString(digest[:]), Length: int64(len(data)),172 })173 entries = append(entries, preparedEntry{name: named.Name, data: data, method: zip.Store})174 }175 for _, blob := range bundle.Blobs {176 hash, length, err := blobDigest(blob)177 if err != nil {178 return bootstrap{}, nil, fmt.Errorf("prepare %s: %w", blob.Name, err)179 }180 manifest.Blobs = append(manifest.Blobs, blobReference{181 Name: blob.Name, Hash: hash, Length: length,182 })183 entries = append(entries, preparedEntry{name: blob.Name, data: blob.Data, sourcePath: blob.Path, method: zip.Store})184 }185 slices.SortFunc(manifest.Tables, func(left, right tableReference) int { return strings.Compare(left.Name, right.Name) })186 slices.SortFunc(manifest.Blobs, func(left, right blobReference) int { return strings.Compare(left.Name, right.Name) })187 return manifest, entries, nil188}189190func blobDigest(blob Blob) (string, int64, error) {191 if blob.Path == "" {192 digest := sha256.Sum256(blob.Data)193 return hex.EncodeToString(digest[:]), int64(len(blob.Data)), nil194 }195 if len(blob.Data) != 0 {196 return "", 0, fmt.Errorf("blob has both data and a source path")197 }198 file, err := os.Open(blob.Path)199 if err != nil {200 return "", 0, err201 }202 defer file.Close()203 digest := sha256.New()204 length, err := io.Copy(digest, file)205 if err != nil {206 return "", 0, err207 }208 return hex.EncodeToString(digest.Sum(nil)), length, nil209}210211func writeArchive(out io.Writer, entries []preparedEntry) error {212 archive := zip.NewWriter(out)213 seen := make(map[string]bool, len(entries))214 for _, entry := range entries {215 if err := validEntryName(entry.name); err != nil {216 return err217 }218 if seen[entry.name] {219 return fmt.Errorf("entry %s is written twice", entry.name)220 }221 seen[entry.name] = true222 header := &zip.FileHeader{Name: entry.name, Method: entry.method}223 header.SetModTime(time.Date(1980, 1, 1, 0, 0, 0, 0, time.UTC))224 writer, err := archive.CreateHeader(header)225 if err != nil {226 return fmt.Errorf("create %s: %w", entry.name, err)227 }228 if entry.sourcePath != "" {229 source, err := os.Open(entry.sourcePath)230 if err != nil {231 return fmt.Errorf("open %s: %w", entry.name, err)232 }233 _, copyErr := io.Copy(writer, source)234 closeErr := source.Close()235 if copyErr != nil {236 return fmt.Errorf("write %s: %w", entry.name, copyErr)237 }238 if closeErr != nil {239 return fmt.Errorf("close %s: %w", entry.name, closeErr)240 }241 continue242 }243 if _, err := writer.Write(entry.data); err != nil {244 return fmt.Errorf("write %s: %w", entry.name, err)245 }246 }247 if err := archive.Close(); err != nil {248 return fmt.Errorf("close Atlas archive: %w", err)249 }250 return nil251}252253// Reader is an opened vNext bundle. It retains both the file schema and the254// application's merged runtime schema.255type Reader struct {256 VolumeID string257 Release Release258 FileSchema Schema259 RuntimeSchema Schema260261 entries map[string]*zip.File262 tables map[string]tableReference263 blobs map[string]blobReference264 limits Limits265 source io.ReaderAt266}267268// Open reads the bootstrap and embedded schema. Only container framing is a269// hard gate; schema differences are merged by identity.270func Open(source io.ReaderAt, size int64, applicationSchema Schema) (*Reader, error) {271 return OpenWithLimits(source, size, applicationSchema, DefaultLimits())272}273274// OpenWithLimits opens a native Atlas under an explicit reproducible resource275// policy. Open is the compatibility facade using DefaultLimits.276func OpenWithLimits(source io.ReaderAt, size int64, applicationSchema Schema, limits Limits) (*Reader, error) {277 limits, err := limits.resolved()278 if err != nil {279 return nil, err280 }281 archive, err := zip.NewReader(source, size)282 if err != nil {283 return nil, fmt.Errorf("open Atlas archive: %w", err)284 }285 entries, err := indexEntries(archive.File, limits)286 if err != nil {287 return nil, err288 }289 manifest, err := readBootstrap(entries)290 if err != nil {291 return nil, err292 }293 fileSchema, err := readSchema(entries, manifest.Schema)294 if err != nil {295 return nil, err296 }297 runtimeSchema, err := UnionSchemas(applicationSchema, fileSchema)298 if err != nil {299 return nil, fmt.Errorf("merge runtime schema: %w", err)300 }301 reader := &Reader{302 VolumeID: manifest.Volume, Release: manifest.Release, FileSchema: fileSchema, RuntimeSchema: runtimeSchema,303 entries: entries, tables: make(map[string]tableReference, len(manifest.Tables)),304 blobs: make(map[string]blobReference, len(manifest.Blobs)), limits: limits, source: source,305 }306 if err := reader.indexReferences(manifest); err != nil {307 return nil, err308 }309 return reader, nil310}311312func indexEntries(files []*zip.File, limits Limits) (map[string]*zip.File, error) {313 if len(files) > limits.MaxEntries {314 return nil, fmt.Errorf("archive carries %d entries, limit is %d", len(files), limits.MaxEntries)315 }316 entries := make(map[string]*zip.File, len(files))317 var total uint64318 for _, file := range files {319 if err := validEntryName(file.Name); err != nil {320 return nil, err321 }322 if _, held := entries[file.Name]; held {323 return nil, fmt.Errorf("archive carries %s twice", file.Name)324 }325 if file.UncompressedSize64 > math.MaxInt64 || total > math.MaxUint64-file.UncompressedSize64 {326 return nil, fmt.Errorf("archive entry sizes overflow")327 }328 total += file.UncompressedSize64329 if total > uint64(limits.MaxTotalBytes) {330 return nil, fmt.Errorf("archive expands beyond %d bytes", limits.MaxTotalBytes)331 }332 entries[file.Name] = file333 }334 return entries, nil335}336337func readBootstrap(entries map[string]*zip.File) (bootstrap, error) {338 data, err := readLimitedEntry(entries, manifestName, maxManifestSize)339 if err != nil {340 return bootstrap{}, err341 }342 var manifest bootstrap343 decoder := json.NewDecoder(bytes.NewReader(data))344 decoder.DisallowUnknownFields()345 if err := decoder.Decode(&manifest); err != nil {346 return bootstrap{}, fmt.Errorf("decode bootstrap: %w", err)347 }348 if err := decoder.Decode(&struct{}{}); err != io.EOF {349 return bootstrap{}, fmt.Errorf("bootstrap carries trailing data")350 }351 if manifest.Format != containerFormat || manifest.Framing < minimumFraming || manifest.Framing > containerFraming {352 return bootstrap{}, fmt.Errorf("unsupported Atlas container identity or framing")353 }354 if err := ValidSlug(manifest.Volume); err != nil {355 return bootstrap{}, fmt.Errorf("volume identity: %w", err)356 }357 if err := manifest.Release.Validate(); err != nil {358 return bootstrap{}, fmt.Errorf("release: %w", err)359 }360 stamp := manifest.Release.Stamp361 manifest.Release.Stamp = ""362 stampInput, err := json.Marshal(manifest)363 if err != nil {364 return bootstrap{}, fmt.Errorf("marshal stamp input: %w", err)365 }366 digest := sha256.Sum256(stampInput)367 if hex.EncodeToString(digest[:]) != stamp {368 return bootstrap{}, fmt.Errorf("release stamp does not match bundle contents")369 }370 manifest.Release.Stamp = stamp371 return manifest, nil372}373374func readSchema(entries map[string]*zip.File, reference schemaReference) (Schema, error) {375 data, err := readLimitedEntry(entries, reference.Name, 16<<20)376 if err != nil {377 return Schema{}, err378 }379 digest := sha256.Sum256(data)380 if hex.EncodeToString(digest[:]) != reference.Hash {381 return Schema{}, fmt.Errorf("embedded schema hash does not match")382 }383 var schema Schema384 if err := json.Unmarshal(data, &schema); err != nil {385 return Schema{}, fmt.Errorf("decode embedded schema: %w", err)386 }387 canonical, err := schema.Canonical()388 if err != nil {389 return Schema{}, err390 }391 if !bytes.Equal(canonical, data) {392 return Schema{}, fmt.Errorf("embedded schema is not canonical")393 }394 return schema, nil395}396397func (reader *Reader) indexReferences(manifest bootstrap) error {398 if len(manifest.Tables) > reader.limits.MaxTables || len(manifest.Blobs) > reader.limits.MaxBlobs {399 return fmt.Errorf("native container reference count exceeds policy")400 }401 allowed := map[string]bool{manifestName: true, manifest.Schema.Name: true}402 for _, reference := range manifest.Tables {403 entry := reader.entries[reference.Name]404 if allowed[reference.Name] || entry == nil {405 return fmt.Errorf("table reference %s is duplicate or missing", reference.Name)406 }407 if err := validateReference(reference.Name, reference.Hash, reference.Length, reader.limits.MaxTableBytes, entry); err != nil {408 return fmt.Errorf("table reference %s: %w", reference.Name, err)409 }410 reader.tables[reference.Name] = reference411 allowed[reference.Name] = true412 }413 for _, reference := range manifest.Blobs {414 entry := reader.entries[reference.Name]415 if allowed[reference.Name] || entry == nil {416 return fmt.Errorf("blob reference %s is duplicate or missing", reference.Name)417 }418 if err := validateReference(reference.Name, reference.Hash, reference.Length, reader.limits.MaxBlobBytes, entry); err != nil {419 return fmt.Errorf("blob reference %s: %w", reference.Name, err)420 }421 reader.blobs[reference.Name] = reference422 allowed[reference.Name] = true423 }424 for name := range reader.entries {425 if !allowed[name] {426 return fmt.Errorf("archive carries unreferenced entry %s", name)427 }428 }429 return nil430}431432func validateReference(name, hash string, length, maximum int64, entry *zip.File) error {433 if name == manifestName || length < 0 || length > maximum || int64(entry.UncompressedSize64) != length {434 return fmt.Errorf("length is invalid or exceeds %d bytes", maximum)435 }436 decoded, err := hex.DecodeString(hash)437 if err != nil || len(decoded) != sha256.Size || hash != strings.ToLower(hash) {438 return fmt.Errorf("hash is not lowercase SHA-256")439 }440 if entry.Method != zip.Store {441 return fmt.Errorf("referenced packed data is not stored")442 }443 return nil444}445446// Table decodes and integrity-checks one named table.447func (reader *Reader) Table(name string) (Table, error) {448 reference, held := reader.tables[name]449 if !held {450 if table, partitioned, err := reader.partitionedTable(name); partitioned || err != nil {451 return table, err452 }453 return Table{}, fmt.Errorf("bundle has no table %s", name)454 }455 data, err := readReferencedEntry(reader.entries[name], reference.Length, reference.Hash, reader.limits.MaxTableBytes)456 if err != nil {457 return Table{}, err458 }459 block, err := DecodeBlock(data)460 if err != nil {461 return Table{}, fmt.Errorf("decode %s: %w", name, err)462 }463 schemaHash, err := reader.FileSchema.Hash()464 if err != nil {465 return Table{}, err466 }467 if block.SchemaHash != schemaHash || block.Table.TypeID != reference.RootType {468 return Table{}, fmt.Errorf("table %s disagrees with its schema or root type", name)469 }470 fileType, held := reader.FileSchema.Type(block.Table.TypeID)471 if !held {472 return Table{}, fmt.Errorf("schema has no root type for table %s", name)473 }474 for index := range block.Table.Columns {475 field, held := fileType.Field(block.Table.Columns[index].Field.ID)476 if !held || field.Kind != block.Table.Columns[index].Field.Kind {477 return Table{}, fmt.Errorf("schema has no matching field for table %s", name)478 }479 block.Table.Columns[index].Field = field480 }481 for _, field := range fileType.Fields {482 if field.Optional {483 continue484 }485 if !tableHasColumn(block.Table, field.ID) {486 return Table{}, fmt.Errorf("table %s omits required field %s", name, field.ID)487 }488 }489 return block.Table, nil490}491492// TableBlock returns the verified physical bytes for browser or range-based493// consumers that decode columns themselves.494func (reader *Reader) TableBlock(name string) ([]byte, error) {495 reference, held := reader.tables[name]496 if !held {497 if table, partitioned, err := reader.partitionedTable(name); partitioned || err != nil {498 if err != nil {499 return nil, err500 }501 schemaHash, err := reader.FileSchema.Hash()502 if err != nil {503 return nil, err504 }505 return EncodeBlock(schemaHash, table)506 }507 return nil, fmt.Errorf("bundle has no table %s", name)508 }509 return readReferencedEntry(reader.entries[name], reference.Length, reference.Hash, reader.limits.MaxTableBytes)510}511512func tableHasColumn(table Table, fieldID ID) bool {513 for _, column := range table.Columns {514 if column.Field.ID == fieldID {515 return true516 }517 }518 return false519}520521func (reader *Reader) Blob(name string) ([]byte, error) {522 reference, held := reader.blobs[name]523 if !held {524 if tile, err := reader.ReadRasterTile(name); err == nil {525 return tile.Data, nil526 }527 return nil, fmt.Errorf("bundle has no blob %s", name)528 }529 return readReferencedEntry(reader.entries[name], reference.Length, reference.Hash, reader.limits.MaxBlobBytes)530}531532func (reader *Reader) TableNames() []string {533 names := make([]string, 0, len(reader.tables))534 for name := range reader.tables {535 names = append(names, name)536 }537 slices.Sort(names)538 return names539}540541func (reader *Reader) BlobNames() []string {542 names, _ := reader.LogicalBlobNames()543 return names544}545546// LogicalBlobNames returns public asset and tile names. Unlike BlobNames it547// reports a malformed logical index instead of degrading to a partial list.548func (reader *Reader) LogicalBlobNames() ([]string, error) {549 names := make([]string, 0, len(reader.blobs))550 for name := range reader.blobs {551 if name != rasterIndexName && name != featureIndexName && !strings.HasPrefix(name, rasterShardPrefix) {552 names = append(names, name)553 }554 }555 index, err := reader.rasterIndex()556 if err != nil {557 return nil, err558 }559 for _, tile := range index.Tiles {560 names = append(names, tile.Name)561 }562 slices.Sort(names)563 return names, nil564}565566// Validate reads and verifies every typed table and opaque blob, then restores567// the semantic root. Installers call it before and after committing a file.568func (reader *Reader) Validate() error {569 featureIndex, err := reader.featureIndex()570 if err != nil {571 return fmt.Errorf("validate feature index: %w", err)572 }573 if _, err := reader.rasterIndex(); err != nil {574 return fmt.Errorf("validate raster index: %w", err)575 }576 for _, name := range reader.TableNames() {577 table, err := reader.Table(name)578 if err != nil {579 return fmt.Errorf("validate %s: %w", name, err)580 }581 for _, partition := range featureIndex.Partitions {582 if partition.Features == name && table.Rows != partition.Rows {583 return fmt.Errorf("validate %s: index row count does not match", name)584 }585 }586 }587 blobNames := make([]string, 0, len(reader.blobs))588 for name := range reader.blobs {589 blobNames = append(blobNames, name)590 }591 slices.Sort(blobNames)592 for _, name := range blobNames {593 reference := reader.blobs[name]594 if err := validateReferencedEntry(reader.entries[name], reference.Length, reference.Hash, reader.limits.MaxBlobBytes); err != nil {595 return fmt.Errorf("validate %s: %w", name, err)596 }597 }598 if _, err := reader.Volume(); err != nil {599 return fmt.Errorf("restore volume: %w", err)600 }601 return nil602}603604func readLimitedEntry(entries map[string]*zip.File, name string, maximum int64) ([]byte, error) {605 entry := entries[name]606 if entry == nil || int64(entry.UncompressedSize64) > maximum {607 return nil, fmt.Errorf("entry %s is missing or implausibly large", name)608 }609 return readEntryLimited(entry, maximum)610}611612func readReferencedEntry(entry *zip.File, length int64, hash string, maximum int64) ([]byte, error) {613 if entry == nil || length < 0 || length > maximum || int64(entry.UncompressedSize64) != length {614 return nil, fmt.Errorf("referenced entry length does not match")615 }616 data, err := readEntryLimited(entry, maximum)617 if err != nil {618 return nil, err619 }620 digest := sha256.Sum256(data)621 if hex.EncodeToString(digest[:]) != hash {622 return nil, fmt.Errorf("referenced entry hash does not match")623 }624 return data, nil625}626627func validateReferencedEntry(entry *zip.File, length int64, hash string, maximum int64) error {628 if entry == nil || length < 0 || length > maximum || int64(entry.UncompressedSize64) != length {629 return fmt.Errorf("referenced entry length does not match")630 }631 source, err := entry.Open()632 if err != nil {633 return fmt.Errorf("open %s: %w", entry.Name, err)634 }635 defer source.Close()636 digest := sha256.New()637 written, err := io.Copy(digest, io.LimitReader(source, plusOne(maximum)))638 if err != nil {639 return fmt.Errorf("read %s: %w", entry.Name, err)640 }641 if written != length || hex.EncodeToString(digest.Sum(nil)) != hash {642 return fmt.Errorf("referenced entry length or hash does not match")643 }644 return nil645}646647func readEntry(entry *zip.File) ([]byte, error) {648 return readEntryLimited(entry, math.MaxInt64-1)649}650651func readEntryLimited(entry *zip.File, maximum int64) ([]byte, error) {652 source, err := entry.Open()653 if err != nil {654 return nil, fmt.Errorf("open %s: %w", entry.Name, err)655 }656 defer source.Close()657 data, err := io.ReadAll(io.LimitReader(source, plusOne(maximum)))658 if err != nil {659 return nil, fmt.Errorf("read %s: %w", entry.Name, err)660 }661 if int64(len(data)) > maximum {662 return nil, fmt.Errorf("entry %s expands beyond %d bytes", entry.Name, maximum)663 }664 return data, nil665}666667func plusOne(value int64) int64 {668 if value == math.MaxInt64 {669 return value670 }671 return value + 1672}673674func validEntryName(name string) error {675 if name == "" || strings.HasPrefix(name, "/") || strings.HasSuffix(name, "/") {676 return fmt.Errorf("entry name %q is not relative", name)677 }678 for segment := range strings.SplitSeq(name, "/") {679 if segment == "" || segment == "." || segment == ".." {680 return fmt.Errorf("entry name %q climbs or stutters", name)681 }682 }683 return nil684}