306 lines
8.3 KiB
Go
306 lines
8.3 KiB
Go
package http
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"math"
|
|
"math/rand"
|
|
"net"
|
|
"net/http"
|
|
"net/url"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/debridmediamanager/zurg/internal/config"
|
|
"github.com/debridmediamanager/zurg/pkg/hosts"
|
|
"github.com/debridmediamanager/zurg/pkg/logutil"
|
|
http_dialer "github.com/mwitkow/go-http-dialer"
|
|
"golang.org/x/net/proxy"
|
|
|
|
cmap "github.com/orcaman/concurrent-map/v2"
|
|
)
|
|
|
|
type HTTPClient struct {
|
|
client *http.Client
|
|
maxRetries int
|
|
timeoutSecs int
|
|
backoff func(attempt int) time.Duration
|
|
bearerToken string
|
|
ensureIPv6Host bool
|
|
cfg config.ConfigInterface
|
|
ipv6 cmap.ConcurrentMap[string, string]
|
|
ipv6Hosts []string
|
|
log *logutil.Logger
|
|
}
|
|
|
|
// {
|
|
// "error": "infringing_file",
|
|
// "error_code": 35
|
|
// }
|
|
|
|
type ApiErrorResponse struct {
|
|
Message string `json:"error"`
|
|
Code int `json:"error_code"`
|
|
}
|
|
|
|
func (e *ApiErrorResponse) Error() string {
|
|
return fmt.Sprintf("api response error: %s (code: %d)", e.Message, e.Code)
|
|
}
|
|
|
|
func NewHTTPClient(token string, maxRetries int, timeoutSecs int, ensureIPv6Host bool, cfg config.ConfigInterface, log *logutil.Logger) *HTTPClient {
|
|
client := HTTPClient{
|
|
bearerToken: token,
|
|
client: http.DefaultClient,
|
|
maxRetries: maxRetries,
|
|
timeoutSecs: timeoutSecs,
|
|
backoff: backoffFunc,
|
|
ensureIPv6Host: ensureIPv6Host,
|
|
cfg: cfg,
|
|
ipv6: cmap.New[string](),
|
|
log: log,
|
|
}
|
|
|
|
var dialer proxy.Dialer = &net.Dialer{
|
|
Timeout: time.Duration(timeoutSecs) * time.Second, // timeout to establish connection
|
|
}
|
|
if proxyURLString := cfg.GetProxy(); proxyURLString != "" {
|
|
proxyURL, err := url.Parse(proxyURLString)
|
|
if err != nil {
|
|
log.Errorf("Failed to parse proxy URL: %v", err)
|
|
return nil
|
|
}
|
|
dialer, err = client.proxyDialer(proxyURL)
|
|
if err != nil {
|
|
log.Errorf("Failed to create proxy dialer: %v", err)
|
|
return nil
|
|
}
|
|
}
|
|
|
|
maxConnections := cfg.GetNumOfWorkers()
|
|
if maxConnections > 32 {
|
|
maxConnections = 32
|
|
}
|
|
|
|
if cfg.ShouldForceIPv6() {
|
|
ipv6List, err := hosts.FetchHosts(hosts.IPV6)
|
|
if err != nil {
|
|
log.Warnf("Failed to fetch IPv6 hosts: %v", err)
|
|
// Decide if you want to return nil here or continue without IPv6
|
|
} else {
|
|
client.ipv6Hosts = ipv6List
|
|
log.Debugf("Fetched %d IPv6 hosts", len(ipv6List))
|
|
}
|
|
|
|
client.client.Transport = &http.Transport{
|
|
ResponseHeaderTimeout: time.Duration(timeoutSecs) * time.Second,
|
|
MaxIdleConns: 0,
|
|
MaxConnsPerHost: maxConnections,
|
|
DialContext: func(ctx context.Context, network, address string) (net.Conn, error) {
|
|
if ipv6Address, ok := client.ipv6.Get(address); ok {
|
|
return dialer.Dial(network, ipv6Address)
|
|
}
|
|
host, port, err := net.SplitHostPort(address)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
ips, err := net.DefaultResolver.LookupIPAddr(ctx, host)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, ip := range ips {
|
|
if ip.IP.To4() == nil { // IPv6 address found
|
|
ipv6Address := net.JoinHostPort(ip.IP.String(), port)
|
|
client.ipv6.Set(address, ipv6Address)
|
|
return dialer.Dial(network, ipv6Address)
|
|
}
|
|
}
|
|
return dialer.Dial(network, address)
|
|
},
|
|
}
|
|
} else {
|
|
client.client.Transport = &http.Transport{
|
|
ResponseHeaderTimeout: time.Duration(timeoutSecs) * time.Second,
|
|
MaxIdleConns: 0,
|
|
MaxConnsPerHost: maxConnections,
|
|
DialContext: func(ctx context.Context, network, address string) (net.Conn, error) {
|
|
return dialer.Dial(network, address)
|
|
},
|
|
}
|
|
}
|
|
|
|
return &client
|
|
}
|
|
|
|
func (r *HTTPClient) Do(req *http.Request) (*http.Response, error) {
|
|
if r.bearerToken != "" {
|
|
req.Header.Set("Authorization", "Bearer "+r.bearerToken)
|
|
}
|
|
// check if Range header is set
|
|
reqHasRangeHeader := req.Header.Get("Range") != "" && !strings.HasPrefix(req.Header.Get("Range"), "bytes=0-")
|
|
|
|
var resp *http.Response
|
|
var err error
|
|
attempt := 0
|
|
|
|
for {
|
|
if resp != nil && resp.Body != nil {
|
|
resp.Body.Close()
|
|
}
|
|
r.replaceHostIfNeeded(req) // needed for ipv6
|
|
resp, err = r.client.Do(req)
|
|
if resp != nil && resp.StatusCode/100 >= 4 {
|
|
body, _ := io.ReadAll(resp.Body)
|
|
if body != nil {
|
|
var errResp ApiErrorResponse
|
|
jsonErr := json.Unmarshal(body, &errResp)
|
|
if jsonErr == nil {
|
|
errResp.Message += fmt.Sprintf(" (status code: %d)", resp.StatusCode)
|
|
err = &errResp
|
|
}
|
|
}
|
|
}
|
|
incr := r.shouldRetry(resp, reqHasRangeHeader, err, r.cfg.GetRateLimitSleepSecs())
|
|
if incr > 0 {
|
|
attempt += incr
|
|
if attempt > r.maxRetries {
|
|
break
|
|
}
|
|
time.Sleep(r.backoff(attempt))
|
|
} else if incr == 0 {
|
|
time.Sleep(10 * time.Millisecond)
|
|
} else {
|
|
// don't retry anymore
|
|
break
|
|
}
|
|
}
|
|
return resp, err
|
|
}
|
|
|
|
func (r *HTTPClient) replaceHostIfNeeded(req *http.Request) {
|
|
if strings.HasSuffix(req.Host, "download.real-debrid.cloud") || !r.ensureIPv6Host || !r.cfg.ShouldForceIPv6() {
|
|
return
|
|
}
|
|
// get subdomain of req.Host
|
|
subdomain := strings.Split(req.Host, ".")[0]
|
|
// check if subdomain is numeric
|
|
_, err := strconv.Atoi(subdomain)
|
|
if err == nil {
|
|
// subdomain is numeric, replace it with .cloud
|
|
req.Host = strings.Replace(req.Host, ".com", ".cloud", 1)
|
|
req.URL.Host = req.Host
|
|
}
|
|
// check if host is in the list of IPv6 hosts
|
|
found := false
|
|
for _, h := range r.ipv6Hosts {
|
|
if h == req.Host {
|
|
found = true
|
|
break
|
|
}
|
|
}
|
|
if !found && !r.CanFetchFirstByte(req.URL.String()) {
|
|
req.Host = r.ipv6Hosts[rand.Intn(len(r.ipv6Hosts))]
|
|
req.URL.Host = req.Host
|
|
r.log.Debugf("Host is not a valid IPv6 host, assigning a random IPv6 host: %s", req.Host)
|
|
} else {
|
|
r.ipv6Hosts = append(r.ipv6Hosts, req.Host)
|
|
}
|
|
}
|
|
|
|
func (r *HTTPClient) proxyDialer(proxyURL *url.URL) (proxy.Dialer, error) {
|
|
if proxyURL.Scheme == "http" || proxyURL.Scheme == "https" {
|
|
httpProxyDialer := http_dialer.New(proxyURL, http_dialer.WithConnectionTimeout(time.Duration(r.timeoutSecs)*time.Second))
|
|
return httpProxyDialer, nil
|
|
} else if proxyURL.Scheme == "socks5" {
|
|
return proxy.FromURL(proxyURL, proxy.Direct)
|
|
}
|
|
return nil, fmt.Errorf("unsupported proxy scheme: %s", proxyURL.Scheme)
|
|
}
|
|
|
|
func (r *HTTPClient) shouldRetry(resp *http.Response, reqHasRangeHeader bool, err error, rateLimitSleep int) int {
|
|
if err != nil && strings.HasPrefix(err.Error(), "api response error:") {
|
|
if apiErr, ok := err.(*ApiErrorResponse); ok {
|
|
switch apiErr.Code {
|
|
case -1: // Internal error
|
|
return 1
|
|
case 5: // Slow down (retry infinitely)
|
|
time.Sleep(time.Duration(rateLimitSleep) * time.Second)
|
|
return 0
|
|
case 6: // Ressource unreachable
|
|
return 1
|
|
case 17: // Hoster in maintenance
|
|
return 1
|
|
case 18: // Hoster limit reached
|
|
return 1
|
|
case 19: // Hoster temporarily unavailable
|
|
return 1
|
|
case 25: // Service unavailable
|
|
return 1
|
|
case 34: // Too many requests (retry infinitely)
|
|
time.Sleep(time.Duration(rateLimitSleep) * time.Second)
|
|
return 0
|
|
case 36: // Fair Usage Limit
|
|
return 1
|
|
default:
|
|
return -1 // don't retry
|
|
}
|
|
}
|
|
}
|
|
if err != nil && strings.Contains(err.Error(), "timeout") {
|
|
if resp.Request.URL.Host == "api.real-debrid.com" && !strings.Contains(resp.Request.URL.Path, "unrestrict/") {
|
|
r.log.Warnf("Request timed out, adjust the timeout value in the config")
|
|
}
|
|
return 1
|
|
}
|
|
if resp != nil {
|
|
if resp.StatusCode == 429 {
|
|
time.Sleep(time.Duration(rateLimitSleep) * time.Second)
|
|
return 0
|
|
}
|
|
if resp.StatusCode/100 == 2 && resp.Header.Get("Content-Range") == "" && reqHasRangeHeader {
|
|
time.Sleep(10 * time.Millisecond)
|
|
return 0
|
|
}
|
|
return -1 // don't retry
|
|
}
|
|
return 1
|
|
}
|
|
|
|
func backoffFunc(attempt int) time.Duration {
|
|
maxDuration := 60
|
|
backoff := int(math.Pow(2, float64(attempt)))
|
|
if backoff > maxDuration {
|
|
backoff = maxDuration
|
|
}
|
|
return time.Duration(backoff) * time.Second
|
|
}
|
|
|
|
func (r *HTTPClient) CanFetchFirstByte(url string) bool {
|
|
timeout := time.Duration(r.timeoutSecs) * time.Second
|
|
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
|
defer cancel()
|
|
req, err := http.NewRequest("GET", url, nil)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
req.Header.Set("Range", "bytes=0-0")
|
|
req = req.WithContext(ctx)
|
|
resp, err := r.client.Do(req)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
// If server supports partial content
|
|
if resp.StatusCode/100 == 2 {
|
|
buffer := make([]byte, 1)
|
|
_, err = resp.Body.Read(buffer)
|
|
if err == nil {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|