conn.go 21 KB

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