[#] lambda_worker: Add twitter OAuth key retrieve job type

Author: nykma

cmd/lambda_worker/main.go

@@ -74,11 +74,18 @@ func handler(ctx context.Context, sqs_event events.SQSEvent) (events.SQSEventRes
 		case types.QueueActions.ArweaveUpload:
 			arweaveMsgs[message.Persona] = raw_message.MessageId
 		case types.QueueActions.Revalidate:
-			if err := revalidate_single(ctx, &message); err != nil {
+			if err := revalidateSingle(ctx, &message); err != nil {
 				fmt.Printf("error revalidating proof record %d: %s\n", message.ProofID, err)
 				// Ignore failed revalidation job since failed job will still update DB.
 				// failures = append(failures, events.SQSBatchItemFailure{ItemIdentifier: raw_message.MessageId})
 			}
+	        case types.QueueActions.TwitterOAuthTokenAcquire: {
+			err := twitterRetrieveOAuthToken()
+			if err != nil {
+				// Ignore errors for now
+				fmt.Printf("Error when retrieving Twitter OAuth key: %s", err.Error())
+			}
+		}
 		default:
 			logrus.Warnf("unsupported queue action: %s", message.Action)
 			failures = append(failures, events.SQSBatchItemFailure{ItemIdentifier: raw_message.MessageId})
@@ -137,7 +144,7 @@ func arweave_upload_many(personas []string) error {
 				break
 			}
 
-			item, err := arweave_bundle_single(pc, previous)
+			item, err := arweaveBundleSingle(pc, previous)
 			if err != nil {
 				logrus.Errorf("error marshalling proof chain %s: %w", pc.Uuid, err)
 				break
@@ -169,7 +176,7 @@ func arweave_upload_many(personas []string) error {
 	return nil
 }
 
-func arweave_bundle_single(pc *model.ProofChain, previous *model.ProofChain) (*artypes.BundleItem, error) {
+func arweaveBundleSingle(pc *model.ProofChain, previous *model.ProofChain) (*artypes.BundleItem, error) {
 	previousUuid := ""
 	previousArweaveID := ""
 	if previous != nil {
@@ -224,7 +231,7 @@ func arweave_bundle_single(pc *model.ProofChain, previous *model.ProofChain) (*a
 	return &item, nil
 }
 
-func revalidate_single(ctx context.Context, message *types.QueueMessage) error {
+func revalidateSingle(ctx context.Context, message *types.QueueMessage) error {
 	proof := model.Proof{}
 	tx := model.DB.Preload("ProofChain").Preload("ProofChain.Previous").Where("id = ?", message.ProofID).First(&proof)
 	if tx.Error != nil {
@@ -233,15 +240,15 @@ func revalidate_single(ctx context.Context, message *types.QueueMessage) error {
 	return proof.Revalidate()
 }
 
-func init_db(cfg aws.Config) {
+func initDB(cfg aws.Config) {
 	model.Init(false) // TODO: should read auto migrate from ENV
 }
 
 // func init_sqs(cfg aws.Config) {
 // 	sqs.Init(cfg)
 // }
 
-func init_validators() {
+func initValidators() {
 	twitter.Init()
 	ethereum.Init()
 	keybase.Init()
@@ -264,19 +271,19 @@ func init() {
 		logrus.Fatalf("Unable to load AWS config: %s", err)
 	}
 	common.CurrentRuntime = common.Runtimes.Lambda
-	init_config_from_aws_secret()
+	initConfigFromAWSSecret()
 	logrus.SetLevel(logrus.InfoLevel)
 
-	init_db(cfg)
+	initDB(cfg)
 	// init_sqs(cfg)
-	init_validators()
+	initValidators()
 }
 
-func init_config_from_aws_secret() {
+func initConfigFromAWSSecret() {
 	if initialized {
 		return
 	}
-	secret_name := getE("SECRET_NAME", "")
+	secretName := getE("SECRET_NAME", "")
 	region := getE("SECRET_REGION", "")
 
 	// Create a Secrets Manager client
@@ -290,7 +297,7 @@ func init_config_from_aws_secret() {
 
 	client := secretsmanager.NewFromConfig(cfg)
 	input := secretsmanager.GetSecretValueInput{
-		SecretId:     aws.String(secret_name),
+		SecretId:     aws.String(secretName),
 		VersionStage: aws.String("AWSCURRENT"),
 	}
 	result, err := client.GetSecretValue(context.Background(), &input)
@@ -319,17 +326,33 @@ func init_config_from_aws_secret() {
 	initialized = true
 }
 
-func getE(env_key, default_value string) string {
-	result := os.Getenv(env_key)
+func getE(envKey, defaultValue string) string {
+	result := os.Getenv(envKey)
 	if len(result) == 0 {
-		if len(default_value) > 0 {
-			return default_value
+		if len(defaultValue) > 0 {
+			return defaultValue
 		} else {
-			logrus.Fatalf("ENV %s must be given! Abort.", env_key)
+			logrus.Fatalf("ENV %s must be given! Abort.", envKey)
 			return ""
 		}
 
 	} else {
 		return result
 	}
 }
+
+func twitterRetrieveOAuthToken() (err error) {
+	type TokenList struct {
+		Tokens [] twitter.Tokens `json:"tokens"`
+	}
+	// TODO: Retrieve existed token from a storage space (i.e., KV / S3)
+
+	tokens, err := twitter.GenerateOauthToken()
+	if err != nil {
+		return err
+	}
+	fmt.Printf("TWITTER OAUTH KEY REGISTERED: %+v", *tokens)
+
+	// TODO: save new token to a storage space (i.e., KV / S3)
+	return nil
+}

types/mq.go

@@ -3,11 +3,13 @@ package types
 type QueueAction string
 
 var QueueActions = struct {
-	Revalidate    QueueAction
-	ArweaveUpload QueueAction
+	Revalidate               QueueAction
+	ArweaveUpload            QueueAction
+	TwitterOAuthTokenAcquire QueueAction
 }{
-	Revalidate:    "revalidate",
-	ArweaveUpload: "arweave_upload",
+	Revalidate:               "revalidate",
+	ArweaveUpload:            "arweave_upload",
+	TwitterOAuthTokenAcquire: "twitter_oauth_token_acquire",
 }
 
 // QueueMessage indicates structure of messages in Amazon SQS.

validator/twitter/api.go

@@ -8,6 +8,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/nextdotid/proof_server/util"
 	"github.com/samber/lo"
 	"golang.org/x/xerrors"
 )
@@ -20,16 +21,18 @@ type APIResponse struct {
 	Text string `json:"text"`
 }
 
-const (
-	BASIC_AUTH_USERNAME = "3rJOl1ODzm9yZy63FACdg"
-	BASIC_AUTH_PASSWORD = "5jPoQ5kQvMJFDYRNE8bQ4rHuds4xJqhvgNJM4awaE8"
-)
+type Tokens struct {
+	AccessToken string `json:"access_token"`
+	GuestToken  string `json:"guest_token"`
+	FlowToken   string `json:"flow_token"`
+	OAuthKey    string `json:"oauth_key"`
+	OAuthSecret string `json:"oauth_secret"`
+	CreatedAt   string `json:"created_at"`
+}
 
-var (
-	// TODO: should save accessToken to somewhere else (shared by all Lambda instances)
-	accessToken string
-	guestToken  string
-	flowToken   string
+const (
+	TWITTER_CONSUMER_KEY = "3rJOl1ODzm9yZy63FACdg"
+	TWITTER_CONSUMER_SECRET = "5jPoQ5kQvMJFDYRNE8bQ4rHuds4xJqhvgNJM4awaE8"
 )
 
 func fetchPostWithAPI(id string, maxRetries int) (tweet *APIResponse, err error) {
@@ -38,7 +41,7 @@ func fetchPostWithAPI(id string, maxRetries int) (tweet *APIResponse, err error)
 	return nil, nil
 }
 
-func setHeaders(req *http.Request, setAccessToken, setGuestToken bool) {
+func setHeaders(req *http.Request, tokens *Tokens, setAccessToken, setGuestToken bool) {
 	req.Header.Set("User-Agent", "TwitterAndroid/9.95.0-release.0 (29950000-r-0) ONEPLUS+A3010/9 (OnePlus;ONEPLUS+A3010;OnePlus;OnePlus3;0;;1;2016)")
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("Accept", "application/json")
@@ -49,20 +52,23 @@ func setHeaders(req *http.Request, setAccessToken, setGuestToken bool) {
 	req.Header.Set("System-User-Agent", "Dalvik/2.1.0 (Linux; U; Android 9; ONEPLUS A3010 Build/PKQ1.181203.001)")
 	req.Header.Set("X-Twitter-Active-User", "yes")
 	if setGuestToken {
-		req.Header.Set("X-Guest-Token", guestToken)
+		req.Header.Set("X-Guest-Token", tokens.GuestToken)
 	}
 	if setAccessToken {
-		req.Header.Set("Authorization", "Bearer "+accessToken)
+		req.Header.Set("Authorization", "Bearer "+tokens.AccessToken)
 	}
 }
 
-func GetOauthToken() (err error) {
-	if flowToken == "" {
-		if err := getFlowToken(); err != nil {
-			return err
-		}
+// GenerateOauthToken generates a new Twitter OAuth guest token
+// which can be used in calling Official APIs.
+func GenerateOauthToken() (tokens *Tokens, err error) {
+	tokens = new(Tokens)
+	tokens.CreatedAt = util.TimeToTimestampString(time.Now())
+
+	if err := tokens.getFlowToken(); err != nil {
+		return nil, err
 	}
-	l.Infof("Access token: %s\nGuest token: %s\nFlow token: %s\n", accessToken, guestToken, flowToken)
+	l.Infof("Access token: %s\nGuest token: %s\nFlow token: %s\n", tokens.AccessToken, tokens.GuestToken, tokens.FlowToken)
 
 	requestBody := fmt.Sprintf(`{
         "flow_token": "%s",
@@ -128,22 +134,22 @@ func GetOauthToken() (err error) {
             "location_permission_prompt": 2,
             "notifications_permission_prompt": 4
         }
-    }`, flowToken)
+    }`, tokens.FlowToken)
 
 	req, err := http.NewRequest("POST", "https://api.twitter.com/1.1/onboarding/task.json", strings.NewReader(requestBody))
 	if err != nil {
-		return err
+		return nil, err
 	}
-	setHeaders(req, true, true)
+	setHeaders(req, tokens, true, true)
 
 	resp, err := new(http.Client).Do(req)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	body, err := io.ReadAll(resp.Body)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	l.Infof("Response: \n%s\n", body)
 	type ResponseSubtask struct {
@@ -163,36 +169,37 @@ func GetOauthToken() (err error) {
 		// Should be "success"
 		Status string `json:"status"`
 		// A new flow token, usually ends with ":3"
-		FlowToken string             `json:"flow_token"`
+		FlowToken string            `json:"flow_token"`
 		Subtasks  []ResponseSubtask `json:"subtasks"`
 	}
 	response := new(Response)
 	err = json.Unmarshal(body, response)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	if response.Errors != nil {
-		return xerrors.Errorf("error when getting oauth token: %+v", *response.Errors)
+		return nil, xerrors.Errorf("error when getting oauth token: %+v", *response.Errors)
 	}
 	if response.Status != "success" {
-		return xerrors.Errorf("wrong API status: %s", response.Status)
+		return nil, xerrors.Errorf("wrong API status: %s", response.Status)
 	}
 
 	st, found := lo.Find(response.Subtasks, func(subtask ResponseSubtask) bool {
 		return (subtask.OpenAccount != nil)
 	})
 	if !found {
-		return xerrors.Errorf("oauth token not found in response")
+		return nil, xerrors.Errorf("oauth token not found in response")
 	}
-	flowToken = response.FlowToken
+	// Update new FlowToken
+	tokens.FlowToken = response.FlowToken
 	l.Infof("OAUTH TOKEN REGISTERED: %s --- %s", st.OpenAccount.OauthToken, st.OpenAccount.OauthTokenSecret)
 
-	return nil
+	return tokens, nil
 }
 
-func getFlowToken() (err error) {
-	if guestToken == "" {
-		if err := getGuestToken(); err != nil {
+func (tokens *Tokens) getFlowToken() (err error) {
+	if tokens.GuestToken == "" {
+		if err := tokens.getGuestToken(); err != nil {
 			return err
 		}
 	}
@@ -269,7 +276,7 @@ func getFlowToken() (err error) {
 	if err != nil {
 		return err
 	}
-	setHeaders(req, true, true)
+	setHeaders(req, tokens, true, true)
 
 	resp, err := new(http.Client).Do(req)
 	if err != nil {
@@ -293,24 +300,24 @@ func getFlowToken() (err error) {
 		return xerrors.Errorf("empty FlowToken")
 	}
 
-	flowToken = response.FlowToken
+	tokens.FlowToken = response.FlowToken
 	return nil
 }
 
-func getGuestToken() (err error) {
-	if guestToken != "" {
+func (tokens *Tokens) getGuestToken() (err error) {
+	if tokens.GuestToken != "" {
 		return nil
 	}
-	if accessToken == "" {
-		if err = getAccessToken(); err != nil {
+	if tokens.AccessToken == "" {
+		if err = tokens.getAccessToken(); err != nil {
 			return err
 		}
 	}
 	req, err := http.NewRequest("POST", "https://api.twitter.com/1.1/guest/activate.json", nil)
 	if err != nil {
 		return err
 	}
-	setHeaders(req, true, false)
+	setHeaders(req, tokens, true, false)
 	type Response struct {
 		GuestToken string `json:"guest_token"`
 	}
@@ -333,13 +340,13 @@ func getGuestToken() (err error) {
 	if response.GuestToken == "" {
 		return xerrors.Errorf("Wrong guest token: %s", response.GuestToken)
 	}
-	guestToken = response.GuestToken
+	tokens.GuestToken = response.GuestToken
 
 	return nil
 }
 
-func getAccessToken() (err error) {
-	if accessToken != "" {
+func (tokens *Tokens) getAccessToken() (err error) {
+	if tokens.AccessToken != "" {
 		return nil
 	}
 
@@ -351,8 +358,8 @@ func getAccessToken() (err error) {
 	if err != nil {
 		return err
 	}
-	setHeaders(req, false, false)
-	req.SetBasicAuth(BASIC_AUTH_USERNAME, BASIC_AUTH_PASSWORD)
+	setHeaders(req, tokens, false, false)
+	req.SetBasicAuth(TWITTER_CONSUMER_KEY, TWITTER_CONSUMER_SECRET)
 	resp, err := new(http.Client).Do(req)
 	if err != nil {
 		return err
@@ -373,6 +380,13 @@ func getAccessToken() (err error) {
 		return xerrors.Errorf("Wrong bearer token: %s %s", response.TokenType, response.AccessToken)
 	}
 
-	accessToken = response.AccessToken
+	tokens.AccessToken = response.AccessToken
 	return nil
 }
+
+func (tokens *Tokens) IsExpired() bool {
+	const EXPIRED_AT = "24h"
+	expiredAt, _ := time.ParseDuration(EXPIRED_AT)
+	createdAt, _ :=  util.TimestampStringToTime(tokens.CreatedAt)
+	return createdAt.Add(expiredAt).Before(time.Now())
+}

validator/twitter/api_test.go

@@ -8,33 +8,38 @@ import (
 
 func Test_getAccessToken(t *testing.T) {
 	t.Run("success", func(t *testing.T) {
-		require.NoError(t, getAccessToken())
-		require.NotEmpty(t, accessToken)
-		t.Logf("Access token: %s", accessToken)
+		tokens := new(Tokens)
+		require.NoError(t, tokens.getAccessToken())
+		require.NotEmpty(t, tokens.AccessToken)
+		t.Logf("Access token: %s", tokens.AccessToken)
 	})
 }
 
 func Test_getGuestToken(t *testing.T) {
 	t.Run("success", func(t *testing.T) {
-		require.NoError(t, getGuestToken())
-		require.NotEmpty(t, guestToken)
-		t.Logf("Guest token: %s", guestToken)
+		tokens := new(Tokens)
+		require.NoError(t, tokens.getGuestToken())
+		require.NotEmpty(t, tokens.GuestToken)
+		t.Logf("Guest token: %s", tokens.GuestToken)
 	})
 }
 
 func Test_FlowToken(t *testing.T) {
 	t.Run("success", func(t *testing.T) {
-		require.NoError(t, getFlowToken())
-		require.NotEmpty(t, flowToken)
-		t.Logf("Flow token: %s", flowToken)
+		tokens := new(Tokens)
+		require.NoError(t, tokens.getFlowToken())
+		require.NotEmpty(t, tokens.FlowToken)
+		t.Logf("Flow token: %s", tokens.FlowToken)
 	})
 }
 
 func Test_OauthToken(t *testing.T) {
 	t.Run("success", func(t *testing.T) {
-		require.NoError(t, GetOauthToken())
-		t.Logf("Access token: %s", accessToken)
-		t.Logf("Guest token: %s", guestToken)
-		t.Logf("Flow token: %s", flowToken)
+		tokens, err := GenerateOauthToken()
+		require.NoError(t, err)
+		require.False(t, tokens.IsExpired())
+		t.Logf("Access token: %s", tokens.AccessToken)
+		t.Logf("Guest token: %s", tokens.GuestToken)
+		t.Logf("Flow token: %s", tokens.FlowToken)
 	})
 }