conn.go 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793
  1. package oss
  2. import (
  3. "bytes"
  4. "crypto/md5"
  5. "encoding/base64"
  6. "encoding/json"
  7. "encoding/xml"
  8. "fmt"
  9. "hash"
  10. "io"
  11. "io/ioutil"
  12. "net"
  13. "net/http"
  14. "net/url"
  15. "os"
  16. "sort"
  17. "strconv"
  18. "strings"
  19. "time"
  20. )
  21. // Conn defines OSS Conn
  22. type Conn struct {
  23. config *Config
  24. url *urlMaker
  25. client *http.Client
  26. }
  27. var signKeyList = []string{"acl", "uploads", "location", "cors",
  28. "logging", "website", "referer", "lifecycle",
  29. "delete", "append", "tagging", "objectMeta",
  30. "uploadId", "partNumber", "security-token",
  31. "position", "img", "style", "styleName",
  32. "replication", "replicationProgress",
  33. "replicationLocation", "cname", "bucketInfo",
  34. "comp", "qos", "live", "status", "vod",
  35. "startTime", "endTime", "symlink",
  36. "x-oss-process", "response-content-type", "x-oss-traffic-limit",
  37. "response-content-language", "response-expires",
  38. "response-cache-control", "response-content-disposition",
  39. "response-content-encoding", "udf", "udfName", "udfImage",
  40. "udfId", "udfImageDesc", "udfApplication", "comp",
  41. "udfApplicationLog", "restore", "callback", "callback-var", "qosInfo",
  42. "policy", "stat", "encryption", "versions", "versioning", "versionId", "requestPayment",
  43. "x-oss-request-payer", "sequential",
  44. "inventory", "inventoryId", "continuation-token", "asyncFetch",
  45. "worm", "wormId", "wormExtend"}
  46. // init initializes Conn
  47. func (conn *Conn) init(config *Config, urlMaker *urlMaker, client *http.Client) error {
  48. if client == nil {
  49. // New transport
  50. transport := newTransport(conn, config)
  51. // Proxy
  52. if conn.config.IsUseProxy {
  53. proxyURL, err := url.Parse(config.ProxyHost)
  54. if err != nil {
  55. return err
  56. }
  57. if config.IsAuthProxy {
  58. if config.ProxyPassword != "" {
  59. proxyURL.User = url.UserPassword(config.ProxyUser, config.ProxyPassword)
  60. } else {
  61. proxyURL.User = url.User(config.ProxyUser)
  62. }
  63. }
  64. transport.Proxy = http.ProxyURL(proxyURL)
  65. }
  66. client = &http.Client{Transport: transport}
  67. if !config.RedirectEnabled {
  68. disableHTTPRedirect(client)
  69. }
  70. }
  71. conn.config = config
  72. conn.url = urlMaker
  73. conn.client = client
  74. return nil
  75. }
  76. // Do sends request and returns the response
  77. func (conn Conn) Do(method, bucketName, objectName string, params map[string]interface{}, headers map[string]string,
  78. data io.Reader, initCRC uint64, listener ProgressListener) (*Response, error) {
  79. urlParams := conn.getURLParams(params)
  80. subResource := conn.getSubResource(params)
  81. uri := conn.url.getURL(bucketName, objectName, urlParams)
  82. resource := conn.getResource(bucketName, objectName, subResource)
  83. return conn.doRequest(method, uri, resource, headers, data, initCRC, listener)
  84. }
  85. // DoURL sends the request with signed URL and returns the response result.
  86. func (conn Conn) DoURL(method HTTPMethod, signedURL string, headers map[string]string,
  87. data io.Reader, initCRC uint64, listener ProgressListener) (*Response, error) {
  88. // Get URI from signedURL
  89. uri, err := url.ParseRequestURI(signedURL)
  90. if err != nil {
  91. return nil, err
  92. }
  93. m := strings.ToUpper(string(method))
  94. req := &http.Request{
  95. Method: m,
  96. URL: uri,
  97. Proto: "HTTP/1.1",
  98. ProtoMajor: 1,
  99. ProtoMinor: 1,
  100. Header: make(http.Header),
  101. Host: uri.Host,
  102. }
  103. tracker := &readerTracker{completedBytes: 0}
  104. fd, crc := conn.handleBody(req, data, initCRC, listener, tracker)
  105. if fd != nil {
  106. defer func() {
  107. fd.Close()
  108. os.Remove(fd.Name())
  109. }()
  110. }
  111. if conn.config.IsAuthProxy {
  112. auth := conn.config.ProxyUser + ":" + conn.config.ProxyPassword
  113. basic := "Basic " + base64.StdEncoding.EncodeToString([]byte(auth))
  114. req.Header.Set("Proxy-Authorization", basic)
  115. }
  116. req.Header.Set(HTTPHeaderHost, req.Host)
  117. req.Header.Set(HTTPHeaderUserAgent, conn.config.UserAgent)
  118. if headers != nil {
  119. for k, v := range headers {
  120. req.Header.Set(k, v)
  121. }
  122. }
  123. // Transfer started
  124. event := newProgressEvent(TransferStartedEvent, 0, req.ContentLength, 0)
  125. publishProgress(listener, event)
  126. if conn.config.LogLevel >= Debug {
  127. conn.LoggerHTTPReq(req)
  128. }
  129. resp, err := conn.client.Do(req)
  130. if err != nil {
  131. // Transfer failed
  132. event = newProgressEvent(TransferFailedEvent, tracker.completedBytes, req.ContentLength, 0)
  133. publishProgress(listener, event)
  134. conn.config.WriteLog(Debug, "[Resp:%p]http error:%s\n", req, err.Error())
  135. return nil, err
  136. }
  137. if conn.config.LogLevel >= Debug {
  138. //print out http resp
  139. conn.LoggerHTTPResp(req, resp)
  140. }
  141. // Transfer completed
  142. event = newProgressEvent(TransferCompletedEvent, tracker.completedBytes, req.ContentLength, 0)
  143. publishProgress(listener, event)
  144. return conn.handleResponse(resp, crc)
  145. }
  146. func (conn Conn) getURLParams(params map[string]interface{}) string {
  147. // Sort
  148. keys := make([]string, 0, len(params))
  149. for k := range params {
  150. keys = append(keys, k)
  151. }
  152. sort.Strings(keys)
  153. // Serialize
  154. var buf bytes.Buffer
  155. for _, k := range keys {
  156. if buf.Len() > 0 {
  157. buf.WriteByte('&')
  158. }
  159. buf.WriteString(url.QueryEscape(k))
  160. if params[k] != nil {
  161. buf.WriteString("=" + strings.Replace(url.QueryEscape(params[k].(string)), "+", "%20", -1))
  162. }
  163. }
  164. return buf.String()
  165. }
  166. func (conn Conn) getSubResource(params map[string]interface{}) string {
  167. // Sort
  168. keys := make([]string, 0, len(params))
  169. signParams := make(map[string]string)
  170. for k := range params {
  171. if conn.config.AuthVersion == AuthV2 {
  172. encodedKey := url.QueryEscape(k)
  173. keys = append(keys, encodedKey)
  174. if params[k] != nil && params[k] != "" {
  175. signParams[encodedKey] = strings.Replace(url.QueryEscape(params[k].(string)), "+", "%20", -1)
  176. }
  177. } else if conn.isParamSign(k) {
  178. keys = append(keys, k)
  179. if params[k] != nil {
  180. signParams[k] = params[k].(string)
  181. }
  182. }
  183. }
  184. sort.Strings(keys)
  185. // Serialize
  186. var buf bytes.Buffer
  187. for _, k := range keys {
  188. if buf.Len() > 0 {
  189. buf.WriteByte('&')
  190. }
  191. buf.WriteString(k)
  192. if _, ok := signParams[k]; ok {
  193. buf.WriteString("=" + signParams[k])
  194. }
  195. }
  196. return buf.String()
  197. }
  198. func (conn Conn) isParamSign(paramKey string) bool {
  199. for _, k := range signKeyList {
  200. if paramKey == k {
  201. return true
  202. }
  203. }
  204. return false
  205. }
  206. // getResource gets canonicalized resource
  207. func (conn Conn) getResource(bucketName, objectName, subResource string) string {
  208. if subResource != "" {
  209. subResource = "?" + subResource
  210. }
  211. if bucketName == "" {
  212. if conn.config.AuthVersion == AuthV2 {
  213. return url.QueryEscape("/") + subResource
  214. }
  215. return fmt.Sprintf("/%s%s", bucketName, subResource)
  216. }
  217. if conn.config.AuthVersion == AuthV2 {
  218. return url.QueryEscape("/"+bucketName+"/") + strings.Replace(url.QueryEscape(objectName), "+", "%20", -1) + subResource
  219. }
  220. return fmt.Sprintf("/%s/%s%s", bucketName, objectName, subResource)
  221. }
  222. func (conn Conn) doRequest(method string, uri *url.URL, canonicalizedResource string, headers map[string]string,
  223. data io.Reader, initCRC uint64, listener ProgressListener) (*Response, error) {
  224. method = strings.ToUpper(method)
  225. req := &http.Request{
  226. Method: method,
  227. URL: uri,
  228. Proto: "HTTP/1.1",
  229. ProtoMajor: 1,
  230. ProtoMinor: 1,
  231. Header: make(http.Header),
  232. Host: uri.Host,
  233. }
  234. tracker := &readerTracker{completedBytes: 0}
  235. fd, crc := conn.handleBody(req, data, initCRC, listener, tracker)
  236. if fd != nil {
  237. defer func() {
  238. fd.Close()
  239. os.Remove(fd.Name())
  240. }()
  241. }
  242. if conn.config.IsAuthProxy {
  243. auth := conn.config.ProxyUser + ":" + conn.config.ProxyPassword
  244. basic := "Basic " + base64.StdEncoding.EncodeToString([]byte(auth))
  245. req.Header.Set("Proxy-Authorization", basic)
  246. }
  247. date := time.Now().UTC().Format(http.TimeFormat)
  248. req.Header.Set(HTTPHeaderDate, date)
  249. req.Header.Set(HTTPHeaderHost, req.Host)
  250. req.Header.Set(HTTPHeaderUserAgent, conn.config.UserAgent)
  251. akIf := conn.config.GetCredentials()
  252. if akIf.GetSecurityToken() != "" {
  253. req.Header.Set(HTTPHeaderOssSecurityToken, akIf.GetSecurityToken())
  254. }
  255. if headers != nil {
  256. for k, v := range headers {
  257. req.Header.Set(k, v)
  258. }
  259. }
  260. conn.signHeader(req, canonicalizedResource)
  261. // Transfer started
  262. event := newProgressEvent(TransferStartedEvent, 0, req.ContentLength, 0)
  263. publishProgress(listener, event)
  264. if conn.config.LogLevel >= Debug {
  265. conn.LoggerHTTPReq(req)
  266. }
  267. resp, err := conn.client.Do(req)
  268. if err != nil {
  269. // Transfer failed
  270. event = newProgressEvent(TransferFailedEvent, tracker.completedBytes, req.ContentLength, 0)
  271. publishProgress(listener, event)
  272. conn.config.WriteLog(Debug, "[Resp:%p]http error:%s\n", req, err.Error())
  273. return nil, err
  274. }
  275. if conn.config.LogLevel >= Debug {
  276. //print out http resp
  277. conn.LoggerHTTPResp(req, resp)
  278. }
  279. // Transfer completed
  280. event = newProgressEvent(TransferCompletedEvent, tracker.completedBytes, req.ContentLength, 0)
  281. publishProgress(listener, event)
  282. return conn.handleResponse(resp, crc)
  283. }
  284. func (conn Conn) signURL(method HTTPMethod, bucketName, objectName string, expiration int64, params map[string]interface{}, headers map[string]string) string {
  285. akIf := conn.config.GetCredentials()
  286. if akIf.GetSecurityToken() != "" {
  287. params[HTTPParamSecurityToken] = akIf.GetSecurityToken()
  288. }
  289. m := strings.ToUpper(string(method))
  290. req := &http.Request{
  291. Method: m,
  292. Header: make(http.Header),
  293. }
  294. if conn.config.IsAuthProxy {
  295. auth := conn.config.ProxyUser + ":" + conn.config.ProxyPassword
  296. basic := "Basic " + base64.StdEncoding.EncodeToString([]byte(auth))
  297. req.Header.Set("Proxy-Authorization", basic)
  298. }
  299. req.Header.Set(HTTPHeaderDate, strconv.FormatInt(expiration, 10))
  300. req.Header.Set(HTTPHeaderUserAgent, conn.config.UserAgent)
  301. if headers != nil {
  302. for k, v := range headers {
  303. req.Header.Set(k, v)
  304. }
  305. }
  306. if conn.config.AuthVersion == AuthV2 {
  307. params[HTTPParamSignatureVersion] = "OSS2"
  308. params[HTTPParamExpiresV2] = strconv.FormatInt(expiration, 10)
  309. params[HTTPParamAccessKeyIDV2] = conn.config.AccessKeyID
  310. additionalList, _ := conn.getAdditionalHeaderKeys(req)
  311. if len(additionalList) > 0 {
  312. params[HTTPParamAdditionalHeadersV2] = strings.Join(additionalList, ";")
  313. }
  314. }
  315. subResource := conn.getSubResource(params)
  316. canonicalizedResource := conn.getResource(bucketName, objectName, subResource)
  317. signedStr := conn.getSignedStr(req, canonicalizedResource, akIf.GetAccessKeySecret())
  318. if conn.config.AuthVersion == AuthV1 {
  319. params[HTTPParamExpires] = strconv.FormatInt(expiration, 10)
  320. params[HTTPParamAccessKeyID] = akIf.GetAccessKeyID()
  321. params[HTTPParamSignature] = signedStr
  322. } else if conn.config.AuthVersion == AuthV2 {
  323. params[HTTPParamSignatureV2] = signedStr
  324. }
  325. urlParams := conn.getURLParams(params)
  326. return conn.url.getSignURL(bucketName, objectName, urlParams)
  327. }
  328. func (conn Conn) signRtmpURL(bucketName, channelName, playlistName string, expiration int64) string {
  329. params := map[string]interface{}{}
  330. if playlistName != "" {
  331. params[HTTPParamPlaylistName] = playlistName
  332. }
  333. expireStr := strconv.FormatInt(expiration, 10)
  334. params[HTTPParamExpires] = expireStr
  335. akIf := conn.config.GetCredentials()
  336. if akIf.GetAccessKeyID() != "" {
  337. params[HTTPParamAccessKeyID] = akIf.GetAccessKeyID()
  338. if akIf.GetSecurityToken() != "" {
  339. params[HTTPParamSecurityToken] = akIf.GetSecurityToken()
  340. }
  341. signedStr := conn.getRtmpSignedStr(bucketName, channelName, playlistName, expiration, akIf.GetAccessKeySecret(), params)
  342. params[HTTPParamSignature] = signedStr
  343. }
  344. urlParams := conn.getURLParams(params)
  345. return conn.url.getSignRtmpURL(bucketName, channelName, urlParams)
  346. }
  347. // handleBody handles request body
  348. func (conn Conn) handleBody(req *http.Request, body io.Reader, initCRC uint64,
  349. listener ProgressListener, tracker *readerTracker) (*os.File, hash.Hash64) {
  350. var file *os.File
  351. var crc hash.Hash64
  352. reader := body
  353. readerLen, err := GetReaderLen(reader)
  354. if err == nil {
  355. req.ContentLength = readerLen
  356. }
  357. req.Header.Set(HTTPHeaderContentLength, strconv.FormatInt(req.ContentLength, 10))
  358. // MD5
  359. if body != nil && conn.config.IsEnableMD5 && req.Header.Get(HTTPHeaderContentMD5) == "" {
  360. md5 := ""
  361. reader, md5, file, _ = calcMD5(body, req.ContentLength, conn.config.MD5Threshold)
  362. req.Header.Set(HTTPHeaderContentMD5, md5)
  363. }
  364. // CRC
  365. if reader != nil && conn.config.IsEnableCRC {
  366. crc = NewCRC(CrcTable(), initCRC)
  367. reader = TeeReader(reader, crc, req.ContentLength, listener, tracker)
  368. }
  369. // HTTP body
  370. rc, ok := reader.(io.ReadCloser)
  371. if !ok && reader != nil {
  372. rc = ioutil.NopCloser(reader)
  373. }
  374. if conn.isUploadLimitReq(req) {
  375. limitReader := &LimitSpeedReader{
  376. reader: rc,
  377. ossLimiter: conn.config.UploadLimiter,
  378. }
  379. req.Body = limitReader
  380. } else {
  381. req.Body = rc
  382. }
  383. return file, crc
  384. }
  385. // isUploadLimitReq: judge limit upload speed or not
  386. func (conn Conn) isUploadLimitReq(req *http.Request) bool {
  387. if conn.config.UploadLimitSpeed == 0 || conn.config.UploadLimiter == nil {
  388. return false
  389. }
  390. if req.Method != "GET" && req.Method != "DELETE" && req.Method != "HEAD" {
  391. if req.ContentLength > 0 {
  392. return true
  393. }
  394. }
  395. return false
  396. }
  397. func tryGetFileSize(f *os.File) int64 {
  398. fInfo, _ := f.Stat()
  399. return fInfo.Size()
  400. }
  401. // handleResponse handles response
  402. func (conn Conn) handleResponse(resp *http.Response, crc hash.Hash64) (*Response, error) {
  403. var cliCRC uint64
  404. var srvCRC uint64
  405. statusCode := resp.StatusCode
  406. if statusCode >= 400 && statusCode <= 505 {
  407. // 4xx and 5xx indicate that the operation has error occurred
  408. var respBody []byte
  409. respBody, err := readResponseBody(resp)
  410. if err != nil {
  411. return nil, err
  412. }
  413. if len(respBody) == 0 {
  414. err = ServiceError{
  415. StatusCode: statusCode,
  416. RequestID: resp.Header.Get(HTTPHeaderOssRequestID),
  417. }
  418. } else {
  419. // Response contains storage service error object, unmarshal
  420. srvErr, errIn := serviceErrFromXML(respBody, resp.StatusCode,
  421. resp.Header.Get(HTTPHeaderOssRequestID))
  422. if errIn != nil { // error unmarshaling the error response
  423. err = fmt.Errorf("oss: service returned invalid response body, status = %s, RequestId = %s", resp.Status, resp.Header.Get(HTTPHeaderOssRequestID))
  424. } else {
  425. err = srvErr
  426. }
  427. }
  428. return &Response{
  429. StatusCode: resp.StatusCode,
  430. Headers: resp.Header,
  431. Body: ioutil.NopCloser(bytes.NewReader(respBody)), // restore the body
  432. }, err
  433. } else if statusCode >= 300 && statusCode <= 307 {
  434. // OSS use 3xx, but response has no body
  435. err := fmt.Errorf("oss: service returned %d,%s", resp.StatusCode, resp.Status)
  436. return &Response{
  437. StatusCode: resp.StatusCode,
  438. Headers: resp.Header,
  439. Body: resp.Body,
  440. }, err
  441. }
  442. if conn.config.IsEnableCRC && crc != nil {
  443. cliCRC = crc.Sum64()
  444. }
  445. srvCRC, _ = strconv.ParseUint(resp.Header.Get(HTTPHeaderOssCRC64), 10, 64)
  446. // 2xx, successful
  447. return &Response{
  448. StatusCode: resp.StatusCode,
  449. Headers: resp.Header,
  450. Body: resp.Body,
  451. ClientCRC: cliCRC,
  452. ServerCRC: srvCRC,
  453. }, nil
  454. }
  455. // LoggerHTTPReq Print the header information of the http request
  456. func (conn Conn) LoggerHTTPReq(req *http.Request) {
  457. var logBuffer bytes.Buffer
  458. logBuffer.WriteString(fmt.Sprintf("[Req:%p]Method:%s\t", req, req.Method))
  459. logBuffer.WriteString(fmt.Sprintf("Host:%s\t", req.URL.Host))
  460. logBuffer.WriteString(fmt.Sprintf("Path:%s\t", req.URL.Path))
  461. logBuffer.WriteString(fmt.Sprintf("Query:%s\t", req.URL.RawQuery))
  462. logBuffer.WriteString(fmt.Sprintf("Header info:"))
  463. for k, v := range req.Header {
  464. var valueBuffer bytes.Buffer
  465. for j := 0; j < len(v); j++ {
  466. if j > 0 {
  467. valueBuffer.WriteString(" ")
  468. }
  469. valueBuffer.WriteString(v[j])
  470. }
  471. logBuffer.WriteString(fmt.Sprintf("\t%s:%s", k, valueBuffer.String()))
  472. }
  473. conn.config.WriteLog(Debug, "%s\n", logBuffer.String())
  474. }
  475. // LoggerHTTPResp Print Response to http request
  476. func (conn Conn) LoggerHTTPResp(req *http.Request, resp *http.Response) {
  477. var logBuffer bytes.Buffer
  478. logBuffer.WriteString(fmt.Sprintf("[Resp:%p]StatusCode:%d\t", req, resp.StatusCode))
  479. logBuffer.WriteString(fmt.Sprintf("Header info:"))
  480. for k, v := range resp.Header {
  481. var valueBuffer bytes.Buffer
  482. for j := 0; j < len(v); j++ {
  483. if j > 0 {
  484. valueBuffer.WriteString(" ")
  485. }
  486. valueBuffer.WriteString(v[j])
  487. }
  488. logBuffer.WriteString(fmt.Sprintf("\t%s:%s", k, valueBuffer.String()))
  489. }
  490. conn.config.WriteLog(Debug, "%s\n", logBuffer.String())
  491. }
  492. func calcMD5(body io.Reader, contentLen, md5Threshold int64) (reader io.Reader, b64 string, tempFile *os.File, err error) {
  493. if contentLen == 0 || contentLen > md5Threshold {
  494. // Huge body, use temporary file
  495. tempFile, err = ioutil.TempFile(os.TempDir(), TempFilePrefix)
  496. if tempFile != nil {
  497. io.Copy(tempFile, body)
  498. tempFile.Seek(0, os.SEEK_SET)
  499. md5 := md5.New()
  500. io.Copy(md5, tempFile)
  501. sum := md5.Sum(nil)
  502. b64 = base64.StdEncoding.EncodeToString(sum[:])
  503. tempFile.Seek(0, os.SEEK_SET)
  504. reader = tempFile
  505. }
  506. } else {
  507. // Small body, use memory
  508. buf, _ := ioutil.ReadAll(body)
  509. sum := md5.Sum(buf)
  510. b64 = base64.StdEncoding.EncodeToString(sum[:])
  511. reader = bytes.NewReader(buf)
  512. }
  513. return
  514. }
  515. func readResponseBody(resp *http.Response) ([]byte, error) {
  516. defer resp.Body.Close()
  517. out, err := ioutil.ReadAll(resp.Body)
  518. if err == io.EOF {
  519. err = nil
  520. }
  521. return out, err
  522. }
  523. func serviceErrFromXML(body []byte, statusCode int, requestID string) (ServiceError, error) {
  524. var storageErr ServiceError
  525. if err := xml.Unmarshal(body, &storageErr); err != nil {
  526. return storageErr, err
  527. }
  528. storageErr.StatusCode = statusCode
  529. storageErr.RequestID = requestID
  530. storageErr.RawMessage = string(body)
  531. return storageErr, nil
  532. }
  533. func xmlUnmarshal(body io.Reader, v interface{}) error {
  534. data, err := ioutil.ReadAll(body)
  535. if err != nil {
  536. return err
  537. }
  538. return xml.Unmarshal(data, v)
  539. }
  540. func jsonUnmarshal(body io.Reader, v interface{}) error {
  541. data, err := ioutil.ReadAll(body)
  542. if err != nil {
  543. return err
  544. }
  545. return json.Unmarshal(data, v)
  546. }
  547. // timeoutConn handles HTTP timeout
  548. type timeoutConn struct {
  549. conn net.Conn
  550. timeout time.Duration
  551. longTimeout time.Duration
  552. }
  553. func newTimeoutConn(conn net.Conn, timeout time.Duration, longTimeout time.Duration) *timeoutConn {
  554. conn.SetReadDeadline(time.Now().Add(longTimeout))
  555. return &timeoutConn{
  556. conn: conn,
  557. timeout: timeout,
  558. longTimeout: longTimeout,
  559. }
  560. }
  561. func (c *timeoutConn) Read(b []byte) (n int, err error) {
  562. c.SetReadDeadline(time.Now().Add(c.timeout))
  563. n, err = c.conn.Read(b)
  564. c.SetReadDeadline(time.Now().Add(c.longTimeout))
  565. return n, err
  566. }
  567. func (c *timeoutConn) Write(b []byte) (n int, err error) {
  568. c.SetWriteDeadline(time.Now().Add(c.timeout))
  569. n, err = c.conn.Write(b)
  570. c.SetReadDeadline(time.Now().Add(c.longTimeout))
  571. return n, err
  572. }
  573. func (c *timeoutConn) Close() error {
  574. return c.conn.Close()
  575. }
  576. func (c *timeoutConn) LocalAddr() net.Addr {
  577. return c.conn.LocalAddr()
  578. }
  579. func (c *timeoutConn) RemoteAddr() net.Addr {
  580. return c.conn.RemoteAddr()
  581. }
  582. func (c *timeoutConn) SetDeadline(t time.Time) error {
  583. return c.conn.SetDeadline(t)
  584. }
  585. func (c *timeoutConn) SetReadDeadline(t time.Time) error {
  586. return c.conn.SetReadDeadline(t)
  587. }
  588. func (c *timeoutConn) SetWriteDeadline(t time.Time) error {
  589. return c.conn.SetWriteDeadline(t)
  590. }
  591. // UrlMaker builds URL and resource
  592. const (
  593. urlTypeCname = 1
  594. urlTypeIP = 2
  595. urlTypeAliyun = 3
  596. )
  597. type urlMaker struct {
  598. Scheme string // HTTP or HTTPS
  599. NetLoc string // Host or IP
  600. Type int // 1 CNAME, 2 IP, 3 ALIYUN
  601. IsProxy bool // Proxy
  602. }
  603. // Init parses endpoint
  604. func (um *urlMaker) Init(endpoint string, isCname bool, isProxy bool) error {
  605. if strings.HasPrefix(endpoint, "http://") {
  606. um.Scheme = "http"
  607. um.NetLoc = endpoint[len("http://"):]
  608. } else if strings.HasPrefix(endpoint, "https://") {
  609. um.Scheme = "https"
  610. um.NetLoc = endpoint[len("https://"):]
  611. } else {
  612. um.Scheme = "http"
  613. um.NetLoc = endpoint
  614. }
  615. //use url.Parse() to get real host
  616. strUrl := um.Scheme + "://" + um.NetLoc
  617. url, err := url.Parse(strUrl)
  618. if err != nil {
  619. return err
  620. }
  621. um.NetLoc = url.Host
  622. host, _, err := net.SplitHostPort(um.NetLoc)
  623. if err != nil {
  624. host = um.NetLoc
  625. if host[0] == '[' && host[len(host)-1] == ']' {
  626. host = host[1 : len(host)-1]
  627. }
  628. }
  629. ip := net.ParseIP(host)
  630. if ip != nil {
  631. um.Type = urlTypeIP
  632. } else if isCname {
  633. um.Type = urlTypeCname
  634. } else {
  635. um.Type = urlTypeAliyun
  636. }
  637. um.IsProxy = isProxy
  638. return nil
  639. }
  640. // getURL gets URL
  641. func (um urlMaker) getURL(bucket, object, params string) *url.URL {
  642. host, path := um.buildURL(bucket, object)
  643. addr := ""
  644. if params == "" {
  645. addr = fmt.Sprintf("%s://%s%s", um.Scheme, host, path)
  646. } else {
  647. addr = fmt.Sprintf("%s://%s%s?%s", um.Scheme, host, path, params)
  648. }
  649. uri, _ := url.ParseRequestURI(addr)
  650. return uri
  651. }
  652. // getSignURL gets sign URL
  653. func (um urlMaker) getSignURL(bucket, object, params string) string {
  654. host, path := um.buildURL(bucket, object)
  655. return fmt.Sprintf("%s://%s%s?%s", um.Scheme, host, path, params)
  656. }
  657. // getSignRtmpURL Build Sign Rtmp URL
  658. func (um urlMaker) getSignRtmpURL(bucket, channelName, params string) string {
  659. host, path := um.buildURL(bucket, "live")
  660. channelName = url.QueryEscape(channelName)
  661. channelName = strings.Replace(channelName, "+", "%20", -1)
  662. return fmt.Sprintf("rtmp://%s%s/%s?%s", host, path, channelName, params)
  663. }
  664. // buildURL builds URL
  665. func (um urlMaker) buildURL(bucket, object string) (string, string) {
  666. var host = ""
  667. var path = ""
  668. object = url.QueryEscape(object)
  669. object = strings.Replace(object, "+", "%20", -1)
  670. if um.Type == urlTypeCname {
  671. host = um.NetLoc
  672. path = "/" + object
  673. } else if um.Type == urlTypeIP {
  674. if bucket == "" {
  675. host = um.NetLoc
  676. path = "/"
  677. } else {
  678. host = um.NetLoc
  679. path = fmt.Sprintf("/%s/%s", bucket, object)
  680. }
  681. } else {
  682. if bucket == "" {
  683. host = um.NetLoc
  684. path = "/"
  685. } else {
  686. host = bucket + "." + um.NetLoc
  687. path = "/" + object
  688. }
  689. }
  690. return host, path
  691. }