oficina / home Public

Download this file

1	package operator
2
3	import (
4	"context"
5	"crypto/hmac"
6	"crypto/sha256"
7	"database/sql"
8	"database/sql/driver"
9	"embed"
10	"errors"
11	"io/fs"
12	"math"
13	"time"
14
15	"amadan.net/rastrillo/rastrillo/migrate"
16	)
17
18	//go:embed migrations
19	var migrations embed.FS
20
21	var CredentialSchema = roleSchema("credential")
22	var DirectorySchema = roleSchema("directory")
23
24	func roleSchema(role string) *migrate.Set {
25	f, e := fs.Sub(migrations, "migrations/"+role)
26	if e != nil {
27	panic(e)
28	}
29	return migrate.MustFromFS(f, "oficina_"+role+"_operator")
30	}
31
32	type Sealer interface {
33	Seal(string, []byte) ([]byte, error)
34	Open(string, []byte) ([]byte, error)
35	}
36
37	// Mutation executes on the engine's already locked connection. It must not
38	// commit, open another writer, dial the network or emit its result itself.
39	type Mutation func(context.Context, *sql.Conn, Request) ([]byte, error)
40
41	type Engine struct {
42	db *sql.DB
43	verifier *Verifier
44	sealer Sealer
45	macKey []byte
46	prefix string
47	now func() time.Time
48	checkSchema func(context.Context, *sql.Conn) error
49	}
50
51	func NewEngine(db *sql.DB, v *Verifier, sealer Sealer, macKey []byte, now func() time.Time) (*Engine, error) {
52	if db == nil || v == nil || sealer == nil || len(macKey) != 32 || now == nil {
53	return nil, ErrRefused
54	}
55	return &Engine{db, v, sealer, append([]byte(nil), macKey...), v.pins.Role + "_operator", now, nil}, nil
56	}
57	func (e *Engine) purpose(a Authorization) string {
58	return "operator.result/v1/" + a.Origin + "/" + a.OperationID + "/" + a.Action
59	}
60	func (e *Engine) mac(r Request) []byte {
61	m := hmac.New(sha256.New, e.macKey)
62	m.Write(r.input)
63	return m.Sum(nil)
64	}
65
66	// Execute validates against this engine's pinned verifier, then commits the
67	// mutation, result and accepted stream together. Expired authentic input is
68	// useful only when its exact previously committed result already exists.
69	func (e *Engine) Execute(ctx context.Context, raw []byte, mutate Mutation) ([]byte, error) {
70	if e == nil || mutate == nil {
71	return nil, ErrRefused
72	}
73	req, err := e.verifier.Verify(raw)
74	if err != nil {
75	return nil, ErrRefused
76	}
77	a := req.authorization
78	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
79	defer cancel()
80	conn, err := e.db.Conn(ctx)
81	if err != nil {
82	return nil, ErrUnavailable
83	}
84	defer conn.Close()
85	for _, statement := range []string{"PRAGMA foreign_keys=ON", "PRAGMA secure_delete=ON", "PRAGMA busy_timeout=1000"} {
86	if _, err = conn.ExecContext(ctx, statement); err != nil {
87	return nil, ErrUnavailable
88	}
89	}
90	var fk, secure int
91	var journal string
92	if conn.QueryRowContext(ctx, "PRAGMA foreign_keys").Scan(&fk) != nil || fk != 1 || conn.QueryRowContext(ctx, "PRAGMA secure_delete").Scan(&secure) != nil || secure != 1 || conn.QueryRowContext(ctx, "PRAGMA journal_mode").Scan(&journal) != nil || journal != "wal" {
93	return nil, ErrUnavailable
94	}
95	violations, err := conn.QueryContext(ctx, "PRAGMA foreign_key_check")
96	if err != nil {
97	return nil, ErrUnavailable
98	}
99	bad := violations.Next()
100	readErr := violations.Err()
101	closeErr := violations.Close()
102	if bad || readErr != nil || closeErr != nil {
103	return nil, ErrUnavailable
104	}
105	if _, err = conn.ExecContext(ctx, "BEGIN IMMEDIATE"); err != nil {
106	return nil, ErrUnavailable
107	}
108	finished := false
109	defer func() {
110	if !finished {
111	cleanup, cancel := context.WithTimeout(context.Background(), time.Second)
112	defer cancel()
113	if _, err := conn.ExecContext(cleanup, "ROLLBACK"); err != nil {
114	_ = conn.Raw(func(any) error { return driver.ErrBadConn })
115	}
116	}
117	}()
118	if e.checkSchema != nil {
119	if err := e.checkSchema(ctx, conn); err != nil {
120	return nil, err
121	}
122	}
123	var storedOrigin, storedInstallation, storedRegistry, storedDevice, storedHead, storedKey string
124	var storedSequence, storedCompleted int64
125	stateErr := conn.QueryRowContext(ctx, "SELECT origin,installation,registry,device,sequence,head,completed_at,key_id FROM "+e.prefix+"_state WHERE singleton=1").Scan(&storedOrigin, &storedInstallation, &storedRegistry, &storedDevice, &storedSequence, &storedHead, &storedCompleted, &storedKey)
126	if stateErr != nil && !errors.Is(stateErr, sql.ErrNoRows) {
127	return nil, ErrUnavailable
128	}
129	if stateErr == nil && (storedOrigin != a.Origin || storedInstallation != a.Installation || storedRegistry != a.Registry || storedDevice != a.Device || storedKey != e.verifier.kid || storedSequence <= 0 || !CanonicalToken(storedHead) || storedCompleted <= 0 || e.now().Unix() < storedCompleted) {
130	return nil, ErrRefused
131	}
132	var oldAction, oldHead, oldOperator string
133	var oldSequence int64
134	var oldMAC, sealed []byte
135	err = conn.QueryRowContext(ctx, "SELECT action,sequence,authorization_head,request_mac,sealed_result,operator_ref FROM "+e.prefix+"_results WHERE operation_id=?", a.OperationID).Scan(&oldAction, &oldSequence, &oldHead, &oldMAC, &sealed, &oldOperator)
136	if err == nil {
137	if stateErr != nil || storedSequence < a.Sequence || oldOperator != a.Device || oldAction != a.Action || oldSequence != a.Sequence || oldHead != req.head || !hmac.Equal(oldMAC, e.mac(req)) {
138	return nil, ErrRefused
139	}
140	result, openErr := e.sealer.Open(e.purpose(a), sealed)
141	if openErr != nil || len(result) == 0 || len(result) > InputLimit {
142	return nil, ErrUnavailable
143	}
144	if _, err = conn.ExecContext(ctx, "COMMIT"); err != nil {
145	return nil, ErrUnavailable
146	}
147	finished = true
148	return result, nil
149	}
150	if !errors.Is(err, sql.ErrNoRows) {
151	return nil, ErrUnavailable
152	}
153	now := e.now()
154	if !req.Fresh(now) {
155	return nil, ErrRefused
156	}
157	var origin, installation, registry, device, head string
158	var sequence, last int64
159	err = conn.QueryRowContext(ctx, "SELECT origin,installation,registry,device,sequence,head,completed_at FROM "+e.prefix+"_state WHERE singleton=1").Scan(&origin, &installation, &registry, &device, &sequence, &head, &last)
160	if errors.Is(err, sql.ErrNoRows) {
161	if !a.Initial || a.Sequence != 1 || a.Previous != "" {
162	return nil, ErrRefused
163	}
164	} else if err != nil {
165	return nil, ErrUnavailable
166	} else {
167	if a.Initial || sequence == math.MaxInt64 || a.Sequence != sequence+1 || a.Previous != head || a.Origin != origin || a.Installation != installation || a.Registry != registry || a.Device != device || now.Unix() < last {
168	return nil, ErrRefused
169	}
170	}
171	var count int
172	if conn.QueryRowContext(ctx, "SELECT count(*) FROM "+e.prefix+"_results").Scan(&count) != nil {
173	return nil, ErrUnavailable
174	}
175	if count >= 10000 {
176	return nil, ErrRefused
177	}
178	result, err := mutate(ctx, conn, req)
179	if err != nil {
180	if errors.Is(err, ErrRefused) {
181	return nil, ErrRefused
182	}
183	return nil, ErrUnavailable
184	}
185	completed := e.now()
186	if !req.Fresh(completed) || completed.Before(now) || len(result) == 0 || len(result) > InputLimit {
187	return nil, ErrRefused
188	}
189	sealed, err = e.sealer.Seal(e.purpose(a), result)
190	if err != nil {
191	return nil, ErrUnavailable
192	}
193	_, err = conn.ExecContext(ctx, "INSERT INTO "+e.prefix+"_results(operation_id,action,sequence,authorization_head,request_mac,sealed_result,completed_at,operator_ref) VALUES(?,?,?,?,?,?,?,?)", a.OperationID, a.Action, a.Sequence, req.head, e.mac(req), sealed, completed.Unix(), a.Device)
194	if err != nil {
195	return nil, ErrUnavailable
196	}
197	_, err = conn.ExecContext(ctx, "INSERT INTO "+e.prefix+"_state(singleton,origin,installation,registry,device,sequence,head,completed_at,key_id) VALUES(1,?,?,?,?,?,?,?,?) ON CONFLICT(singleton) DO UPDATE SET sequence=excluded.sequence,head=excluded.head,completed_at=excluded.completed_at", a.Origin, a.Installation, a.Registry, a.Device, a.Sequence, req.head, completed.Unix(), e.verifier.kid)
198	if err != nil {
199	return nil, ErrUnavailable
200	}
201	if _, err = conn.ExecContext(ctx, "COMMIT"); err != nil {
202	return nil, ErrUnavailable
203	}
204	finished = true
205	return append([]byte(nil), result...), nil
206	}
207