X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/12e80e523f1178c0db49a3c9d856d17bb7855dfe..a7f96d94925721f6fe0be23fc626f00f05be635e:/services/keepstore/handlers.go diff --git a/services/keepstore/handlers.go b/services/keepstore/handlers.go index 1c55bb3d84..a86bb6a5b5 100644 --- a/services/keepstore/handlers.go +++ b/services/keepstore/handlers.go @@ -8,8 +8,8 @@ package main // StatusHandler (GET /status.json) import ( - "bufio" "bytes" + "container/list" "crypto/md5" "encoding/json" "fmt" @@ -21,8 +21,7 @@ import ( "regexp" "runtime" "strconv" - "strings" - "syscall" + "sync" "time" ) @@ -40,24 +39,21 @@ func MakeRESTRouter() *mux.Router { rest.HandleFunc(`/{hash:[0-9a-f]{32}}`, PutBlockHandler).Methods("PUT") rest.HandleFunc(`/{hash:[0-9a-f]{32}}`, DeleteHandler).Methods("DELETE") - - // For IndexHandler we support: - // /index - returns all locators - // /index/{prefix} - returns all locators that begin with {prefix} - // {prefix} is a string of hexadecimal digits between 0 and 32 digits. - // If {prefix} is the empty string, return an index of all locators - // (so /index and /index/ behave identically) - // A client may supply a full 32-digit locator string, in which - // case the server will return an index with either zero or one - // entries. This usage allows a client to check whether a block is - // present, and its size and upload time, without retrieving the - // entire block. - // + // List all blocks stored here. Privileged client only. rest.HandleFunc(`/index`, IndexHandler).Methods("GET", "HEAD") - rest.HandleFunc( - `/index/{prefix:[0-9a-f]{0,32}}`, IndexHandler).Methods("GET", "HEAD") + // List blocks stored here whose hash has the given prefix. + // Privileged client only. + rest.HandleFunc(`/index/{prefix:[0-9a-f]{0,32}}`, IndexHandler).Methods("GET", "HEAD") + + // List volumes: path, device number, bytes used/avail. rest.HandleFunc(`/status.json`, StatusHandler).Methods("GET", "HEAD") + // Replace the current pull queue. + rest.HandleFunc(`/pull`, PullHandler).Methods("PUT") + + // Replace the current trash queue. + rest.HandleFunc(`/trash`, TrashHandler).Methods("PUT") + // Any request which does not match any of these routes gets // 400 Bad Request. rest.NotFoundHandler = http.HandlerFunc(BadRequestHandler) @@ -69,176 +65,106 @@ func BadRequestHandler(w http.ResponseWriter, r *http.Request) { http.Error(w, BadRequestError.Error(), BadRequestError.HTTPCode) } -// FindKeepVolumes scans all mounted volumes on the system for Keep -// volumes, and returns a list of matching paths. -// -// A device is assumed to be a Keep volume if it is a normal or tmpfs -// volume and has a "/keep" directory directly underneath the mount -// point. -// -func FindKeepVolumes() []string { - vols := make([]string, 0) - - if f, err := os.Open(PROC_MOUNTS); err != nil { - log.Fatalf("opening %s: %s\n", PROC_MOUNTS, err) - } else { - scanner := bufio.NewScanner(f) - for scanner.Scan() { - args := strings.Fields(scanner.Text()) - dev, mount := args[0], args[1] - if mount != "/" && - (dev == "tmpfs" || strings.HasPrefix(dev, "/dev/")) { - keep := mount + "/keep" - if st, err := os.Stat(keep); err == nil && st.IsDir() { - vols = append(vols, keep) - } - } - } - if err := scanner.Err(); err != nil { - log.Fatal(err) - } - } - return vols -} - func GetBlockHandler(resp http.ResponseWriter, req *http.Request) { - hash := mux.Vars(req)["hash"] - - log.Printf("%s %s", req.Method, hash) - - hints := mux.Vars(req)["hints"] - - // Parse the locator string and hints from the request. - // TODO(twp): implement a Locator type. - var signature, timestamp string - if hints != "" { - signature_pat, _ := regexp.Compile("^A([[:xdigit:]]+)@([[:xdigit:]]{8})$") - for _, hint := range strings.Split(hints, "+") { - if match, _ := regexp.MatchString("^[[:digit:]]+$", hint); match { - // Server ignores size hints - } else if m := signature_pat.FindStringSubmatch(hint); m != nil { - signature = m[1] - timestamp = m[2] - } else if match, _ := regexp.MatchString("^[[:upper:]]", hint); match { - // Any unknown hint that starts with an uppercase letter is - // presumed to be valid and ignored, to permit forward compatibility. - } else { - // Unknown format; not a valid locator. - http.Error(resp, BadRequestError.Error(), BadRequestError.HTTPCode) - return - } - } - } - - // If permission checking is in effect, verify this - // request's permission signature. if enforce_permissions { - if signature == "" || timestamp == "" { - http.Error(resp, PermissionError.Error(), PermissionError.HTTPCode) - return - } else if IsExpired(timestamp) { - http.Error(resp, ExpiredError.Error(), ExpiredError.HTTPCode) + locator := req.URL.Path[1:] // strip leading slash + if err := VerifySignature(locator, GetApiToken(req)); err != nil { + http.Error(resp, err.Error(), err.(*KeepError).HTTPCode) return - } else { - req_locator := req.URL.Path[1:] // strip leading slash - if !VerifySignature(req_locator, GetApiToken(req)) { - http.Error(resp, PermissionError.Error(), PermissionError.HTTPCode) - return - } } } - block, err := GetBlock(hash) - - // Garbage collect after each GET. Fixes #2865. - // TODO(twp): review Keep memory usage and see if there's - // a better way to do this than blindly garbage collecting - // after every block. - defer runtime.GC() - + block, err := GetBlock(mux.Vars(req)["hash"], false) if err != nil { // This type assertion is safe because the only errors // GetBlock can return are DiskHashError or NotFoundError. - if err == NotFoundError { - log.Printf("%s: not found, giving up\n", hash) - } http.Error(resp, err.Error(), err.(*KeepError).HTTPCode) return } + defer bufs.Put(block) - resp.Header().Set("X-Block-Size", fmt.Sprintf("%d", len(block))) - - _, err = resp.Write(block) - if err != nil { - log.Printf("GetBlockHandler: writing response: %s", err) - } - - return + resp.Header().Set("Content-Length", strconv.Itoa(len(block))) + resp.Header().Set("Content-Type", "application/octet-stream") + resp.Write(block) } func PutBlockHandler(resp http.ResponseWriter, req *http.Request) { - // Garbage collect after each PUT. Fixes #2865. - // See also GetBlockHandler. - defer runtime.GC() - hash := mux.Vars(req)["hash"] - log.Printf("%s %s", req.Method, hash) + // Detect as many error conditions as possible before reading + // the body: avoid transmitting data that will not end up + // being written anyway. + + if req.ContentLength == -1 { + http.Error(resp, SizeRequiredError.Error(), SizeRequiredError.HTTPCode) + return + } - // Read the block data to be stored. - // If the request exceeds BLOCKSIZE bytes, issue a HTTP 500 error. - // if req.ContentLength > BLOCKSIZE { http.Error(resp, TooLongError.Error(), TooLongError.HTTPCode) return } - buf := make([]byte, req.ContentLength) - nread, err := io.ReadFull(req.Body, buf) + if len(KeepVM.AllWritable()) == 0 { + http.Error(resp, FullError.Error(), FullError.HTTPCode) + return + } + + buf := bufs.Get(int(req.ContentLength)) + _, err := io.ReadFull(req.Body, buf) if err != nil { http.Error(resp, err.Error(), 500) - } else if int64(nread) < req.ContentLength { - http.Error(resp, "request truncated", 500) - } else { - if err := PutBlock(buf, hash); err == nil { - // Success; add a size hint, sign the locator if - // possible, and return it to the client. - return_hash := fmt.Sprintf("%s+%d", hash, len(buf)) - api_token := GetApiToken(req) - if PermissionSecret != nil && api_token != "" { - expiry := time.Now().Add(permission_ttl) - return_hash = SignLocator(return_hash, api_token, expiry) - } - resp.Write([]byte(return_hash + "\n")) - } else { - ke := err.(*KeepError) - http.Error(resp, ke.Error(), ke.HTTPCode) - } + bufs.Put(buf) + return } - return + + err = PutBlock(buf, hash) + bufs.Put(buf) + + if err != nil { + ke := err.(*KeepError) + http.Error(resp, ke.Error(), ke.HTTPCode) + return + } + + // Success; add a size hint, sign the locator if possible, and + // return it to the client. + return_hash := fmt.Sprintf("%s+%d", hash, req.ContentLength) + api_token := GetApiToken(req) + if PermissionSecret != nil && api_token != "" { + expiry := time.Now().Add(blob_signature_ttl) + return_hash = SignLocator(return_hash, api_token, expiry) + } + resp.Write([]byte(return_hash + "\n")) } // IndexHandler // A HandleFunc to address /index and /index/{prefix} requests. // func IndexHandler(resp http.ResponseWriter, req *http.Request) { - prefix := mux.Vars(req)["prefix"] - - // Only the data manager may issue /index requests, - // and only if enforce_permissions is enabled. - // All other requests return 403 Forbidden. - api_token := GetApiToken(req) - if !enforce_permissions || - api_token == "" || - data_manager_token != api_token { - http.Error(resp, PermissionError.Error(), PermissionError.HTTPCode) + // Reject unauthorized requests. + if !IsDataManagerToken(GetApiToken(req)) { + http.Error(resp, UnauthorizedError.Error(), UnauthorizedError.HTTPCode) return } - var index string - for _, vol := range KeepVM.Volumes() { - index = index + vol.Index(prefix) + + prefix := mux.Vars(req)["prefix"] + + for _, vol := range KeepVM.AllReadable() { + if err := vol.IndexTo(prefix, resp); err != nil { + // The only errors returned by IndexTo are + // write errors returned by resp.Write(), + // which probably means the client has + // disconnected and this error will never be + // reported to the client -- but it will + // appear in our own error log. + http.Error(resp, err.Error(), http.StatusInternalServerError) + return + } } - resp.Write([]byte(index)) + // An empty line at EOF is the only way the client can be + // assured the entire index was received. + resp.Write([]byte{'\n'}) } // StatusHandler @@ -260,60 +186,66 @@ type VolumeStatus struct { BytesUsed uint64 `json:"bytes_used"` } +type PoolStatus struct { + Alloc uint64 `json:"BytesAllocated"` + Cap int `json:"BuffersMax"` + Len int `json:"BuffersInUse"` +} + type NodeStatus struct { - Volumes []*VolumeStatus `json:"volumes"` + Volumes []*VolumeStatus `json:"volumes"` + BufferPool PoolStatus + PullQueue WorkQueueStatus + TrashQueue WorkQueueStatus + Memory runtime.MemStats } +var st NodeStatus +var stLock sync.Mutex + func StatusHandler(resp http.ResponseWriter, req *http.Request) { - st := GetNodeStatus() - if jstat, err := json.Marshal(st); err == nil { + stLock.Lock() + readNodeStatus(&st) + jstat, err := json.Marshal(&st) + stLock.Unlock() + if err == nil { resp.Write(jstat) } else { log.Printf("json.Marshal: %s\n", err) - log.Printf("NodeStatus = %v\n", st) + log.Printf("NodeStatus = %v\n", &st) http.Error(resp, err.Error(), 500) } } -// GetNodeStatus -// Returns a NodeStatus struct describing this Keep -// node's current status. -// -func GetNodeStatus() *NodeStatus { - st := new(NodeStatus) - - st.Volumes = make([]*VolumeStatus, len(KeepVM.Volumes())) - for i, vol := range KeepVM.Volumes() { - st.Volumes[i] = vol.Status() +// populate the given NodeStatus struct with current values. +func readNodeStatus(st *NodeStatus) { + vols := KeepVM.AllReadable() + if cap(st.Volumes) < len(vols) { + st.Volumes = make([]*VolumeStatus, len(vols)) + } + st.Volumes = st.Volumes[:0] + for _, vol := range vols { + if s := vol.Status(); s != nil { + st.Volumes = append(st.Volumes, s) + } } - return st + st.BufferPool.Alloc = bufs.Alloc() + st.BufferPool.Cap = bufs.Cap() + st.BufferPool.Len = bufs.Len() + st.PullQueue = getWorkQueueStatus(pullq) + st.TrashQueue = getWorkQueueStatus(trashq) + runtime.ReadMemStats(&st.Memory) } -// GetVolumeStatus -// Returns a VolumeStatus describing the requested volume. -// -func GetVolumeStatus(volume string) *VolumeStatus { - var fs syscall.Statfs_t - var devnum uint64 - - if fi, err := os.Stat(volume); err == nil { - devnum = fi.Sys().(*syscall.Stat_t).Dev - } else { - log.Printf("GetVolumeStatus: os.Stat: %s\n", err) - return nil +// return a WorkQueueStatus for the given queue. If q is nil (which +// should never happen except in test suites), return a zero status +// value instead of crashing. +func getWorkQueueStatus(q *WorkQueue) WorkQueueStatus { + if q == nil { + // This should only happen during tests. + return WorkQueueStatus{} } - - err := syscall.Statfs(volume, &fs) - if err != nil { - log.Printf("GetVolumeStatus: statfs: %s\n", err) - return nil - } - // These calculations match the way df calculates disk usage: - // "free" space is measured by fs.Bavail, but "used" space - // uses fs.Blocks - fs.Bfree. - free := fs.Bavail * uint64(fs.Bsize) - used := (fs.Blocks - fs.Bfree) * uint64(fs.Bsize) - return &VolumeStatus{volume, devnum, free, used} + return q.Status() } // DeleteHandler processes DELETE requests. @@ -345,7 +277,6 @@ func GetVolumeStatus(volume string) *VolumeStatus { // func DeleteHandler(resp http.ResponseWriter, req *http.Request) { hash := mux.Vars(req)["hash"] - log.Printf("%s %s", req.Method, hash) // Confirm that this user is an admin and has a token with unlimited scope. var tok = GetApiToken(req) @@ -359,14 +290,14 @@ func DeleteHandler(resp http.ResponseWriter, req *http.Request) { return } - // Delete copies of this block from all available volumes. Report - // how many blocks were successfully and unsuccessfully - // deleted. + // Delete copies of this block from all available volumes. + // Report how many blocks were successfully deleted, and how + // many were found on writable volumes but not deleted. var result struct { Deleted int `json:"copies_deleted"` Failed int `json:"copies_failed"` } - for _, vol := range KeepVM.Volumes() { + for _, vol := range KeepVM.AllWritable() { if err := vol.Delete(hash); err == nil { result.Deleted++ } else if os.IsNotExist(err) { @@ -397,10 +328,112 @@ func DeleteHandler(resp http.ResponseWriter, req *http.Request) { } } -// GetBlock, PutBlock and TouchBlock implement lower-level code for -// handling blocks by rooting through volumes connected to the local -// machine. Once the handler has determined that system policy permits -// the request, it calls these methods to perform the actual operation. +/* PullHandler processes "PUT /pull" requests for the data manager. + The request body is a JSON message containing a list of pull + requests in the following format: + + [ + { + "locator":"e4d909c290d0fb1ca068ffaddf22cbd0+4985", + "servers":[ + "keep0.qr1hi.arvadosapi.com:25107", + "keep1.qr1hi.arvadosapi.com:25108" + ] + }, + { + "locator":"55ae4d45d2db0793d53f03e805f656e5+658395", + "servers":[ + "10.0.1.5:25107", + "10.0.1.6:25107", + "10.0.1.7:25108" + ] + }, + ... + ] + + Each pull request in the list consists of a block locator string + and an ordered list of servers. Keepstore should try to fetch the + block from each server in turn. + + If the request has not been sent by the Data Manager, return 401 + Unauthorized. + + If the JSON unmarshalling fails, return 400 Bad Request. +*/ + +type PullRequest struct { + Locator string `json:"locator"` + Servers []string `json:"servers"` +} + +func PullHandler(resp http.ResponseWriter, req *http.Request) { + // Reject unauthorized requests. + if !IsDataManagerToken(GetApiToken(req)) { + http.Error(resp, UnauthorizedError.Error(), UnauthorizedError.HTTPCode) + return + } + + // Parse the request body. + var pr []PullRequest + r := json.NewDecoder(req.Body) + if err := r.Decode(&pr); err != nil { + http.Error(resp, err.Error(), BadRequestError.HTTPCode) + return + } + + // We have a properly formatted pull list sent from the data + // manager. Report success and send the list to the pull list + // manager for further handling. + resp.WriteHeader(http.StatusOK) + resp.Write([]byte( + fmt.Sprintf("Received %d pull requests\n", len(pr)))) + + plist := list.New() + for _, p := range pr { + plist.PushBack(p) + } + pullq.ReplaceQueue(plist) +} + +type TrashRequest struct { + Locator string `json:"locator"` + BlockMtime int64 `json:"block_mtime"` +} + +func TrashHandler(resp http.ResponseWriter, req *http.Request) { + // Reject unauthorized requests. + if !IsDataManagerToken(GetApiToken(req)) { + http.Error(resp, UnauthorizedError.Error(), UnauthorizedError.HTTPCode) + return + } + + // Parse the request body. + var trash []TrashRequest + r := json.NewDecoder(req.Body) + if err := r.Decode(&trash); err != nil { + http.Error(resp, err.Error(), BadRequestError.HTTPCode) + return + } + + // We have a properly formatted trash list sent from the data + // manager. Report success and send the list to the trash work + // queue for further handling. + resp.WriteHeader(http.StatusOK) + resp.Write([]byte( + fmt.Sprintf("Received %d trash requests\n", len(trash)))) + + tlist := list.New() + for _, t := range trash { + tlist.PushBack(t) + } + trashq.ReplaceQueue(tlist) +} + +// ============================== +// GetBlock and PutBlock implement lower-level code for handling +// blocks by rooting through volumes connected to the local machine. +// Once the handler has determined that system policy permits the +// request, it calls these methods to perform the actual operation. // // TODO(twp): this code would probably be better located in the // VolumeManager interface. As an abstraction, the VolumeManager @@ -408,48 +441,75 @@ func DeleteHandler(resp http.ResponseWriter, req *http.Request) { // block is stored on, so it should be responsible for figuring out // which volume to check for fetching blocks, storing blocks, etc. -func GetBlock(hash string) ([]byte, error) { +// ============================== +// GetBlock fetches and returns the block identified by "hash". If +// the update_timestamp argument is true, GetBlock also updates the +// block's file modification time (for the sake of PutBlock, which +// must update the file's timestamp when the block already exists). +// +// On success, GetBlock returns a byte slice with the block data, and +// a nil error. +// +// If the block cannot be found on any volume, returns NotFoundError. +// +// If the block found does not have the correct MD5 hash, returns +// DiskHashError. +// + +func GetBlock(hash string, update_timestamp bool) ([]byte, error) { // Attempt to read the requested hash from a keep volume. error_to_caller := NotFoundError - for _, vol := range KeepVM.Volumes() { - if buf, err := vol.Get(hash); err != nil { - // IsNotExist is an expected error and may be ignored. - // (If all volumes report IsNotExist, we return a NotFoundError) - // All other errors should be logged but we continue trying to - // read. - switch { - case os.IsNotExist(err): - continue - default: + var vols []Volume + if update_timestamp { + // Pointless to find the block on an unwritable volume + // because Touch() will fail -- this is as good as + // "not found" for purposes of callers who need to + // update_timestamp. + vols = KeepVM.AllWritable() + } else { + vols = KeepVM.AllReadable() + } + + for _, vol := range vols { + buf, err := vol.Get(hash) + if err != nil { + // IsNotExist is an expected error and may be + // ignored. All other errors are logged. In + // any case we continue trying to read other + // volumes. If all volumes report IsNotExist, + // we return a NotFoundError. + if !os.IsNotExist(err) { log.Printf("GetBlock: reading %s: %s\n", hash, err) } - } else { - // Double check the file checksum. - // - filehash := fmt.Sprintf("%x", md5.Sum(buf)) - if filehash != hash { - // TODO(twp): this condition probably represents a bad disk and - // should raise major alarm bells for an administrator: e.g. - // they should be sent directly to an event manager at high - // priority or logged as urgent problems. - // - log.Printf("%s: checksum mismatch for request %s (actual %s)\n", - vol, hash, filehash) - error_to_caller = DiskHashError - } else { - // Success! - if error_to_caller != NotFoundError { - log.Printf("%s: checksum mismatch for request %s but a good copy was found on another volume and returned\n", - vol, hash) - } - return buf, nil + continue + } + // Check the file checksum. + // + filehash := fmt.Sprintf("%x", md5.Sum(buf)) + if filehash != hash { + // TODO: Try harder to tell a sysadmin about + // this. + log.Printf("%s: checksum mismatch for request %s (actual %s)\n", + vol, hash, filehash) + error_to_caller = DiskHashError + bufs.Put(buf) + continue + } + if error_to_caller == DiskHashError { + log.Printf("%s: checksum mismatch for request %s but a good copy was found on another volume and returned", + vol, hash) + } + if update_timestamp { + if err := vol.Touch(hash); err != nil { + error_to_caller = GenericError + log.Printf("%s: Touch %s failed: %s", + vol, hash, error_to_caller) + bufs.Put(buf) + continue } } - } - - if error_to_caller != NotFoundError { - log.Printf("%s: checksum mismatch, no good copy found\n", hash) + return buf, nil } return nil, error_to_caller } @@ -489,22 +549,17 @@ func PutBlock(block []byte, hash string) error { } // If we already have a block on disk under this identifier, return - // success (but check for MD5 collisions). + // success (but check for MD5 collisions). While fetching the block, + // update its timestamp. // The only errors that GetBlock can return are DiskHashError and NotFoundError. // In either case, we want to write our new (good) block to disk, // so there is nothing special to do if err != nil. - if oldblock, err := GetBlock(hash); err == nil { + // + if oldblock, err := GetBlock(hash, true); err == nil { + defer bufs.Put(oldblock) if bytes.Compare(block, oldblock) == 0 { - // The block already exists; update the timestamp and - // return. - // Note that TouchBlock will fail (and therefore - // so will PutBlock) if the block exists but is found on a - // read-only volume. That is intentional: if the user has - // requested N replicas of a block, we want to be sure that - // there are at least N *writable* replicas, so a block - // that cannot be written to should not count toward the - // replication total. - return TouchBlock(hash) + // The block already exists; return success. + return nil } else { return CollisionError } @@ -512,72 +567,61 @@ func PutBlock(block []byte, hash string) error { // Choose a Keep volume to write to. // If this volume fails, try all of the volumes in order. - vol := KeepVM.Choose() - if err := vol.Put(hash, block); err == nil { - return nil // success! - } else { - allFull := true - for _, vol := range KeepVM.Volumes() { - err := vol.Put(hash, block) - if err == nil { - return nil // success! - } - if err != FullError { - // The volume is not full but the write did not succeed. - // Report the error and continue trying. - allFull = false - log.Printf("%s: Write(%s): %s\n", vol, hash, err) - } + if vol := KeepVM.NextWritable(); vol != nil { + if err := vol.Put(hash, block); err == nil { + return nil // success! } + } - if allFull { - log.Printf("all Keep volumes full") - return FullError - } else { - log.Printf("all Keep volumes failed") - return GenericError - } + writables := KeepVM.AllWritable() + if len(writables) == 0 { + log.Print("No writable volumes.") + return FullError } -} -// TouchBlock finds the block identified by hash and updates its -// filesystem modification time with the current time. -func TouchBlock(hash string) error { - for _, vol := range KeepVM.Volumes() { - err := vol.Touch(hash) - if os.IsNotExist(err) { - continue + allFull := true + for _, vol := range writables { + err := vol.Put(hash, block) + if err == nil { + return nil // success! + } + if err != FullError { + // The volume is not full but the + // write did not succeed. Report the + // error and continue trying. + allFull = false + log.Printf("%s: Write(%s): %s\n", vol, hash, err) } - // either err is nil (meaning success) or some error other - // than "file does not exist" (which we should return upward). - return err } - // If we got here, the block was not found on any volume. - return os.ErrNotExist + + if allFull { + log.Print("All volumes are full.") + return FullError + } else { + // Already logged the non-full errors. + return GenericError + } } +var validLocatorRe = regexp.MustCompile(`^[0-9a-f]{32}$`) + // IsValidLocator // Return true if the specified string is a valid Keep locator. // When Keep is extended to support hash types other than MD5, // this should be updated to cover those as well. // func IsValidLocator(loc string) bool { - match, err := regexp.MatchString(`^[0-9a-f]{32}$`, loc) - if err == nil { - return match - } - log.Printf("IsValidLocator: %s\n", err) - return false + return validLocatorRe.MatchString(loc) } +var authRe = regexp.MustCompile(`^OAuth2\s+(.*)`) + // GetApiToken returns the OAuth2 token from the Authorization // header of a HTTP request, or an empty string if no matching // token is found. func GetApiToken(req *http.Request) string { if auth, ok := req.Header["Authorization"]; ok { - if pat, err := regexp.Compile(`^OAuth2\s+(.*)`); err != nil { - log.Println(err) - } else if match := pat.FindStringSubmatch(auth[0]); match != nil { + if match := authRe.FindStringSubmatch(auth[0]); match != nil { return match[1] } } @@ -604,10 +648,7 @@ func CanDelete(api_token string) bool { } // Blocks may be deleted only when Keep has been configured with a // data manager. - if data_manager_token == "" { - return false - } - if api_token == data_manager_token { + if IsDataManagerToken(api_token) { return true } // TODO(twp): look up api_token with the API server @@ -615,3 +656,9 @@ func CanDelete(api_token string) bool { // has unlimited scope return false } + +// IsDataManagerToken returns true if api_token represents the data +// manager's token. +func IsDataManagerToken(api_token string) bool { + return data_manager_token != "" && api_token == data_manager_token +}