DEV Community: Nestor Gutiérrez

Solving User Consistency Issues with AWS Cognito

Nestor Gutiérrez — Sun, 08 Dec 2024 01:55:04 +0000

A few days ago, I was developing an application to store my records (PRs) while practicing sports such as CrossFit and Weightlifting. Everything seemed to be working correctly in what I called my production environment. However, at some point, a significant issue arose: someone managed to register with the same email address and create two almost identical users username !== Username.

This was a major inconvenience, as the person could log in with both users and would not have their information consistently.

The Problem

After a bit of research (I assumed AWS Cognito prevented this behavior automatically 😔), I discovered that a possible solution was to set the email as a sign-in alias. This required making some changes to the Cognito User Pool.

However, there was a problem: I already had registered users, and Cognito does not allow modifying sign-in aliases in an existing User Pool.

The Solution

Create a New User Pool

The solution was to create a new User Pool. Thus, my User Pool V2 was born (yes, a very original name). Once created, I tested and confirmed that:

It was not possible to register a user with the same email.
The system had to be case-insensitive: a user must always be unique, regardless of uppercase or lowercase.

To handle the latter, I discovered Pre-Signup Triggers. These triggers allow you to run a Lambda Function before completing the signup process. In this function, I implemented the following logic:


import boto3

client = boto3.client('cognito-idp')

def lambda_handler(event, context):
    user_pool_id = event['userPoolId']
    username = event['userName'].lower()  # Normalizes the username
    email = event['request']['userAttributes']['email'].lower()  # Normalizes the email

    # Validate username uniqueness
    response = client.list_users(
        UserPoolId=user_pool_id,
        Filter=f'username = "{username}"'
    )
    if len(response['Users']) > 0:
        raise Exception("The username is already registered.")

    # Validate email uniqueness
    response = client.list_users(
        UserPoolId=user_pool_id,
        Filter=f'email = "{email}"'
    )
    if len(response['Users']) > 0:
        raise Exception("The email is already registered.")

    # Normalize username and email in the event
    event['userName'] = username
    event['request']['userAttributes']['email'] = email

    return event

Convert data to lowercase (username and email).
Check if the user already exists.
Update the event and return it.

With this, we ensure data consistency and prevent duplicate users, regardless of uppercase or lowercase usage.

Deployment with AWS CDK

To deploy this solution, I used AWS CDK with TypeScript. An example configuration might look like this:

import { Construct } from "constructs";
import { addTags } from "../common/utilities/addTag";
import { RetentionDays } from "aws-cdk-lib/aws-logs";
import { StringParameter } from "aws-cdk-lib/aws-ssm";
import { PolicyStatement,Effect } from "aws-cdk-lib/aws-iam";
import { CfnOutput,NestedStack,NestedStackProps} from "aws-cdk-lib";
import { PROJECT_NAME,ENVIRONMENT, COGNITO_CALLBACK_URL } from "../../constants";
import { Code, Function, ILayerVersion, LayerVersion, Runtime } from "aws-cdk-lib/aws-lambda";
import { UserPool,AccountRecovery,VerificationEmailStyle,ClientAttributes,UserPoolClient,OAuthScope } from "aws-cdk-lib/aws-cognito";

export class UserPoolStack extends NestedStack {
  private projectEnv: string;

  public readonly userPoolV2: UserPool;
  public readonly userPoolClientV2: UserPoolClient;

  constructor(scope: Construct,id: string,props?: NestedStackProps) {
    super(scope,id,props);

    const REGION = this.region;

    this.projectEnv = `${PROJECT_NAME}-${ENVIRONMENT}`;

    const callbackURl = (COGNITO_CALLBACK_URL) ? COGNITO_CALLBACK_URL : 'http://localhost:3000';

    this.userPoolV2 = new UserPool(this,  `${this.projectEnv}-user-pool-v2`, {
      selfSignUpEnabled: ENVIRONMENT === 'prod',
      accountRecovery: AccountRecovery.EMAIL_ONLY,
      userVerification: {
        emailStyle: VerificationEmailStyle.CODE,
      },
      autoVerify: {
        email: true,
      },
      standardAttributes: {
        email: {
          required: true,
          mutable: true,
        },
      },
      signInAliases: {
        username:  true,
        email: true,
      }
    });

    const clientWritteAttributes = new ClientAttributes()
      .withStandardAttributes({
        fullname: true,
        email: true,
        nickname: true,
        profilePicture: true,
      });

    const clientReadAttributes = new ClientAttributes()
      .withStandardAttributes({
        emailVerified: true,
        email: true,
      });

    this.userPoolClientV2 = new UserPoolClient(this,`${this.projectEnv}-user-pool-client-v2`, {
      userPool: this.userPoolV2,
      authFlows: {
        userPassword: true,
        userSrp: true,
      },
      generateSecret: true,
      preventUserExistenceErrors: true,
      writeAttributes: clientWritteAttributes,
      readAttributes: clientReadAttributes,
      oAuth: {
        flows: {
          authorizationCodeGrant: true,
        },
        scopes: [OAuthScope.EMAIL,OAuthScope.OPENID,OAuthScope.PROFILE],
        callbackUrls: [`${callbackURl}/auth`],
        logoutUrls: [`${callbackURl}/logout`],
      },
    });

    this.userPoolV2.addDomain(`${PROJECT_NAME}-domain`,{
      cognitoDomain: { domainPrefix: `${this.projectEnv}-v2` }
    });

    new CfnOutput(this,'CognitoUspLoginUrlV2',{
      value: `https://${this.projectEnv}-v2.auth.${REGION}.amazoncognito.com/login?response_type=code&client_id=${this.userPoolClientV2.userPoolClientId}&redirect_uri=${callbackURl}/auth`
    });

    new CfnOutput(this,'CognitoURLV2',{
      value: `https://${this.projectEnv}.auth.${REGION}.amazoncognito.com`
    });

    addTags(this.userPoolV2,'project',`${PROJECT_NAME}`);

    this.createCognitoPreSignUpLambdaFunction();
  }

  private createCognitoPreSignUpLambdaFunction(): void {
    const lambdaFunction = new Function(this, 'cognitoPreSignUp', {
      runtime: Runtime.PYTHON_3_11,
      functionName: 'cognito-pre-sign-up',
      handler: 'src.handler.lambda_handler',
      code: Code.fromAsset('./services/cognito_pre_sign_up'),
      environment: {
        LOG_LEVEL: `${ENVIRONMENT === 'prod' ? 'INFO' : 'DEBUG'}`,
      },
      logRetention: ENVIRONMENT === 'prod' ? RetentionDays.ONE_YEAR : RetentionDays.ONE_DAY,
    });

// least privileges principle
    const lambdaPolicy = new PolicyStatement({
      actions: ['cognito-idp:ListUsers'],
      resources: [this.userPoolV2.userPoolArn],
      effect: Effect.ALLOW,
    });

    lambdaFunction.addToRolePolicy(lambdaPolicy);

    const layerArn = StringParameter.valueForStringParameter(this, `${PROJECT_NAME}-${ENVIRONMENT}-lambda-layer-parameter`);

    const newLayerVersion: ILayerVersion = LayerVersion.fromLayerVersionArn(this, 'lambda-layer-common', layerArn);

    lambdaFunction.addLayers(newLayerVersion);

    lambdaFunction.node.addDependency(this.userPoolV2);

    addTags(lambdaFunction, 'project', `${PROJECT_NAME}`);
  }
}

Migrating Existing Users

Now another question arises: what to do with the users who were already registered in the previous User Pool?

In my case, there were only 5 users, so I chose the simplest solution:

Manually re-invite each one with their new data.
Manually update the userIDs in the DynamoDB table.

Conclusion

With this solution, I managed to solve the user duplication issues and ensure consistent data management. I also learned that:

Setting the email as the sign-in alias is crucial to avoid similar issues.
Cognito’s Pre-Signup Triggers offer great flexibility to customize registration logic.
Planning and testing are essential, especially when there are already users in production.

Now the system is more robust and ready to scale without duplication issues.

If you want to store your records you can join LiftWiz

Making a Youtube Browser with Python & React

Nestor Gutiérrez — Wed, 28 Jul 2021 04:17:37 +0000

Backend

In this series of articles, I'll talk about how to create a youtube browser (Basic) using Python with Flask for backend and React in the frontend, I'll divide all this into many parts

Backend (You are here 📍)
Frontend
Deploy

I created this Youtube Browser as a Technical Challenge for a role in my current job. It was so funny because I wanted to share my experience with you. Maybe I'll complete it by adding Unit Tests, CI/CD, and some features to keep it secure.

Well, here we go with the first part of this "mini-project". Let's create the backend!

In this case, I use one API with Python (Flask) this is built using Docker for this we make a file called Dockerfile with the code below:

FROM python:3.8-alpine

COPY requirements.txt .

RUN apk update

RUN pip install -r requirements.txt

COPY app /app

RUN ls -la

WORKDIR /app

EXPOSE 5000

ENTRYPOINT [ "./start.sh" ]

Now we'll create a file called requirements.txt, this file contains all the dependencies used in this microservice.

Flask==2.0
gunicorn==20.1.0
pendulum==2.1.2
google-api-python-client
google-auth-oauthlib
google-auth-httplib2
oauth2client
webargs==8.0.0

Now we need to create a docker-compose file to start the server on the local environment.

version: "3.6"
services:
    youtube-search:
        container_name: youtube-search-service
        build: .
        image: python:latest
        ports:
            - 5000:5000
        volumes:
            - ./app/app.py:/app/app.py
            - ./app/utils.py:/app/utils.py
            - ./app/models/schema.py:/app/models/schema.py
            - ./app/youtube_manager.py:/app/youtube_manager.py
            - ./app/response_manager.py:/app/response_manager.py
        environment:
            TIMEOUT: 100
            DEBUG: "true"
            LOG_LEVEL: INFO
            TIMEZONE: America/Santiago
            DATE_FORMAT: YYYY-MM-DD HH:mm:ss
            API_KEY_CREDENTIALS: ${API_KEY_CREDENTIALS}

As you can see, in the docker-compose file we have many environment variables one of these is ${API_KEY_CREDENTIALS} this variable is in a .env file and refer to API provided by Google to connect with this API, you can get this on your Google Console Dashboard.

Well now start with the real code, that was just the process to set up the container. We will create a folder called "app" that contains all the files used in this project.

Now create a file called app.py in this one we're going to add the code shown below to consult the Youtube API and return a response to the frontend.

First, we need to import all packages used in this file, some of them will be used in the future. For now add this and make the "check" route work ok.

# app.py

import json
import logging
from os import environ
from flask import Flask
from models.schema import SearchSchema
from webargs.flaskparser import use_args
from googleapiclient.errors import HttpError
from response_manager import ResponseManager
from youtube_manager import YoutubeManager
from marshmallow.exceptions import ValidationError
import utils as utils

Second, we will initialize Flask and set the logger level, additionally, we'll initialize the response manager.

# app.py

# Init Flask
app = Flask(__name__)

# Set logger level
LOG_LEVEL = environ.get('LOG_LEVEL', 'INFO')
logging.basicConfig(level=LOG_LEVEL)

As you can see we have the package response_manager that contains a Class and this helps us (as the name reveals) to manage all the responses we make to the frontend. Let's create it and later instance it in the app.py file.

# response_manager.py

import json
import logging
from os import environ

JSON_MIMETYPE='application/json'
LOG_LEVEL = environ.get('LOG_LEVEL', 'INFO')

# Set logger level
logging.basicConfig(level=LOG_LEVEL)

class ResponseManager():

    def __init__(self, app) -> None:
        self.app = app
        self.mimetype=JSON_MIMETYPE
        self.headers={'Access-Control-Allow-Origin': '*', 'Content-Type': JSON_MIMETYPE}
                # list of all possible errors
        self.list_status = { 400: 'fail', 422: 'fail', 500: 'error', 200:'success', 203: 'success' }
        self.status_code = None

    def message(self, message, status_code) -> dict:
        self.status_code = status_code
        try:
            return self.app.response_class(
                response = json.dumps({
                    'code': self.status_code
                    , 'status': self.list_status.get(self.status_code, 200) # default: 200
                    , 'message': message
                })
                , status=self.status_code
                , mimetype=self.mimetype
                , headers=self.headers
            )
        except Exception as error:
            logging.error(error)
            logging.exception('Something went wrong trying send message')

Now we need an instance of the ResponseManager previously created, add the next code to do it.

# app.py

# Init ResponseManager
response_manager = ResponseManager(app)

Now add a route to check if the service is running ok and everything is fine, then add the next code.

# app.py

@app.route('/')
def index():
    return response_manager.message('Service is Working', 200)

Next, we need to add the final code to make flask RUN, this make run on port 5000 and with the environment variable DEBUG we enable the debug on it.

if __name__ == '__main__':
    app.run(debug=environ.get('DEBUG', "true"), host='0.0.0.0', port=5000)

At this point the app.py file looks like this.

import json
import logging
from os import environ
from flask import Flask
from models.schema import SearchSchema
from webargs.flaskparser import use_args
from googleapiclient.errors import HttpError
from response_manager import ResponseManager
from youtube_manager import YoutubeManager
from marshmallow.exceptions import ValidationError
import utils as utils

# Init Flask
app = Flask(__name__)

# Set logger level
LOG_LEVEL = environ.get('LOG_LEVEL', 'INFO')
logging.basicConfig(level=LOG_LEVEL)

# Init ResponseManager
response_manager = ResponseManager(app)

@app.route('/')
def index():
    return response_manager.message('Service is Working', 200)

if __name__ == '__main__':
    app.run(debug=environ.get('DEBUG', "true"), host='0.0.0.0', port=5000)

Ok now you can test it by making a request using Postman and verify if the container is working, if this is right you can go to the second step and add the route to request the Youtube API, and response to the frontend but first we need to create some utils functions and a YoutubeManager to help us do a "better" code, let's go with the YoutubeManager.

# youtube_manager.py

import logging
from os import environ
from googleapiclient.discovery import build

# Set logger level
LOG_LEVEL = environ.get('LOG_LEVEL', 'INFO')
logging.basicConfig(level=LOG_LEVEL)

class YoutubeManager():

    def __init__(self, api_key) -> None:
        self.developer_api_key = api_key
        self.youtube_service_name = "youtube"
        self.youtube_api_version = "v3"
        self._youtube_cli = None

    def __enter__(self):
        try:
            logging.info('Initializing YouTube Manager')
            if self._youtube_cli is None:
                self._youtube_cli = build(
                    self.youtube_service_name
                    , self.youtube_api_version
                    , developerKey=self.developer_api_key
                )
            else:
                logging.info('Return existent client')
        except Exception as error:
            logging.error(error)
        else:
            logging.info('Returning YouTube Manager')
            return self._youtube_cli

    def __exit__(self, type, value, traceback) -> None:
        try:
            if self._youtube_cli is not None:
                self._youtube_cli = None

        except Exception as error:
            logging.error(error.args)
        else:
            logging.info('YouTube client closed')

So we have our YoutubeManager now let's go and create all utils functions, these functions will help us prepare the response to the frontend, I made it separate to isolate each part of code on small pieces for an easy debug in case of any problem.

# utils.py

# Prepare the response 
def normalize_response(search_response, per_page = 100) -> dict:
    return {
        'prev_page': search_response.get('prevPageToken', None)
        , 'next_page': search_response.get('nextPageToken', None)
        , 'page_info': search_response.get('pageInfo', {'totalResults': 0, 'resultsPerPage': per_page})
        , 'items': normalize_items_response(search_response.get('items', []))
    }

# Prepare each item only with the required fields
def normalize_items_response(items) -> list:
    list_videos = []
    for item in items:
        item_id = item.get('id')
        item_snippet = item.get('snippet')
        item_thumbnails = item_snippet.get('thumbnails')

        new_item = {
            'id': get_id_item(item_id)
            , 'type': get_type_item(item_id.get('kind'))
            , 'description': item_snippet.get('description')
            , 'title': item_snippet.get('title')
            , 'channel_id': item_snippet.get('channelId')
            , 'channel_title': item_snippet.get('channelTitle')
            , 'published_at': item_snippet.get('publishedAt')
            , 'thumbnails': item_thumbnails.get('high')
        }
        list_videos.append(new_item)

    return list_videos

# Validate & Return the type of item
def get_type_item(kind):
    if kind == 'youtube#video':
        return 'video'
    elif kind == 'youtube#channel':
        return 'channel'
    else:
        return 'playlist'

# Validate & Return the ID according to each type
def get_id_item(item):
    if item.get('kind') == 'youtube#video':
        return item.get('videoId')
    elif item.get('kind') == 'youtube#channel':
        return item.get('channelId')
    else:
        return item.get('playlistId')

Well with this complete we can create finally our final function/route to request the Youtube API, add the code below in the app.py .

# app.py

@app.route('/search', methods=['GET'])
def search_in_youtube(args):
    try:
        with YoutubeManager(environ.get('API_KEY_CREDENTIALS')) as youtube_manager:
            logging.info('Initializing search in Youtube API')
            max_results = args.get('per_page', 100)
            # Validating per_page parameter
            if (args.get('per_page') is not None):
                if int(args.get('per_page')) < 0:
                    raise ValidationError('Items per page must be greater than zero (0)')

            # We do the search using the YouTube API
            search_response = youtube_manager.search().list(
                q=args.get('query_search')
                , part='id, snippet'
                , maxResults=max_results
                , pageToken=args.get('page_token', None)
            ).execute()

            response = utils.normalize_response(search_response, max_results)

            return response_manager.message(response, 200)
    except ValidationError as error:
        logging.info(error)
        return response_manager.message(error.args, 422)
        # If is an HttpErrorException make send this message and log the error.
    except HttpError as error:
        error = json.loads(error.args[1])
        logging.error(error)
        return response_manager.message('Something went wrong searching in Youtube API', 500)
    except Exception as error:
        logging.error(error)
        return response_manager.message(error.args, 500)

Wait! But what will happen if someone sends bad parameters in the request?

Well, to solve it we'll add some query validations and respond to the message indicating something is wrong.

# app.py

# Added function to send response when there is an error with status code 422
@app.errorhandler(422)
def validation_error(err):
    messages = err.data.get('messages')
    if 'query' in messages:
        return response_manager.message(messages.get('query'), 422)
    elif 'json' in messages:
        return response_manager.message(messages.get('json'), 422)

Also created the file "models/schema.py" with the code below:

# models/schema.py

from marshmallow import Schema, fields
from marshmallow.exceptions import ValidationError

class SearchSchema(Schema):
    per_page=fields.Integer()
    query_search=fields.String(required=True)
    page_token=fields.String()

Now add the next code @use_args(SearchSchema(), location='query') in the app.py file and the function will look like this.

# app.py

@app.route('/search', methods=['GET'])
@use_args(SearchSchema(), location='query') # -> Added code
def search_in_youtube(args):
    # search function code created before.

At the end, the app.py will looks like this:

import json
import logging
from os import environ
from flask import Flask
from models.schema import SearchSchema
from webargs.flaskparser import use_args
from googleapiclient.errors import HttpError
from response_manager import ResponseManager
from youtube_manager import YoutubeManager
from marshmallow.exceptions import ValidationError
import utils as utils

# Init Flask
app = Flask(__name__)

# Set logger level
LOG_LEVEL = environ.get('LOG_LEVEL', 'INFO')
logging.basicConfig(level=LOG_LEVEL)

# Init ResponseManager
response_manager = ResponseManager(app)

# Added function to send response when there is an error with status code 422
@app.errorhandler(422)
def validation_error(err):
    messages = err.data.get('messages')
    if 'query' in messages:
        return response_manager.message(messages.get('query'), 422)
    elif 'json' in messages:
        return response_manager.message(messages.get('json'), 422)

@app.route('/')
def index():
    return response_manager.message('Service is Working', 200)

@app.route('/search', methods=['GET'])
@use_args(SearchSchema(), location='query')
def search_in_youtube(args):
    try:
        with YoutubeManager(environ.get('API_KEY_CREDENTIALS')) as youtube_manager:
            logging.info('Initializing search in Youtube API')
            max_results = args.get('per_page', 100)
            # Validating per_page parameter
            if (args.get('per_page') is not None):
                if int(args.get('per_page')) < 0:
                    raise ValidationError('Items per page must be greater than zero (0)')

            # We do the search using the YouTube API
            search_response = youtube_manager.search().list(
                q=args.get('query_search')
                , part='id, snippet'
                , maxResults=max_results
                , pageToken=args.get('page_token', None)
            ).execute()

            response = utils.normalize_response(search_response, max_results)

            return response_manager.message(response, 200)
    except ValidationError as error:
        logging.info(error)
        return response_manager.message(error.args, 422)
    except HttpError as error:
        error = json.loads(error.args[1])
        logging.error(error)
        return response_manager.message('Something went wrong searching in Youtube API', 500)
    except Exception as error:
        logging.error(error)
        return response_manager.message(error.args, 500)

if __name__ == '__main__':
    app.run(debug=environ.get('DEBUG', "true"), host='0.0.0.0', port=5000)

Creating a command Cli with Golang

Nestor Gutiérrez — Sat, 22 Aug 2020 03:28:19 +0000

Creating a command Cli with Golang to clone repositories from AWS CodeCommit

My Problem

I’m starting in the AWS world and right now in the project I’m working we need to clone many repositories from AWS CodeCommit, to clone any repo we need to write something like that:

  git clone “https://username:pass@git-codecommit.us-east-1.amazonaws.com/v1/repos/demo-with-super-extense-name

but the username and password need to be URL encoded before cloning the repo, so it was so boring, really I need to do something like this

Url encode the username
Url encode the password
Cut the URL just using the way I need
Create the https://username:pass@git-codecommit.us-east-1.amazonaws.com/v1/repos/demo-with-super-extense-name

Just before cloning a repository? So I decided to create my script to clone all the repositories I want from AWS CodeCommit just writing something like this:

  cloneaws clone —profile=myProfile —url=“ssh-Url-or-https-Url” —projectName=demo

This is more friendly and so easy without URL encode first the username and password

Before start, I had months trying to learn Go, because I feel empathy with the pet of the language (just kidding I just want to learn another language, but the pet is so cute), so I decide to write a command CLI. with Go

Solution

Well, the principal idea is clone a repository from AWS CodeCommit without writing the user and password, just set a label to find in a list of credentials.

First, we create a subcommand with the code below:

  cloneAwsCommand := flag.NewFlagSet("clone", flag.ExitOnError)

Add the few parameters needed by the command with the next code:

  cloneUrl := cloneAwsCommand.String("url", "", "Repository HTTPS/SSH URL (Required)")
  cloneProfile := cloneAwsCommand.String("profile", "", "Profile name to use cloning repository (Required)")
  cloneProjectName := cloneAwsCommand.String("projectName", "", "Folder name to copy repo (optional)")
  cloneHelp := cloneAwsCommand.Bool("help", false, "Show usage for subcommand")

Validate if has set any subcommand:

if len(os.Args) < 2 {
  log.Fatalln("Subcommand is required")
}

Now I'll check what subcommand to use, for now just has clone but I pretend to add many other functionalities like "add a new profile, modified profile"

To check the subcommand we use a switch with the bode below:

switch os.Args[1] {
    case "clone":
        // Clone block
    default:
        fmt.Println("  clone string")
        fmt.Println("   Subcommand for clone")
        log.Fatalln("Subcommand is required")
    }

the switch has a default, this will print in console an error with the message "subcommand is required".

Now I want to explain the code in the clone block:


  // First we parse the arguments for clone subcommand.
  cloneAwsCommand.Parse(os.Args[2:])

  // Validate if set the flag "help" to display the clone subcommand usage.
  if *cloneHelp == true {
      cloneAwsCommand.Usage()
      os.Exit(1)
  }

  /*
    Validate if the profile and URL flag is an empty string.
    If one of them is empty print default values.
  */
  if *cloneProfile == "" || *cloneUrl == "" {
      cloneAwsCommand.PrintDefaults()
      log.Fatalln("No flags set for clone subcommand")
  }

  //Validate again if profile is an empty string.
  if *cloneProfile == "" {
      cloneAwsCommand.PrintDefaults()
      log.Fatalln("No profile value to find")
  }
  /*
    Call findCredentials function to get the username and password.
    I'll explain this function next.
  */
  credentials := findCredentials(*cloneProfile)

  // Validate again if clone URL is and empty string.
  if *cloneUrl == "" {
      cloneAwsCommand.PrintDefaults()
      log.Fatalln("")
  }

  // Get the URL substring.
  urlRepo := getRepoPath(*cloneUrl)

  // URL encode the email and password from credentials.
  userEncode := url.QueryEscape(credentials.Email)
  passwordEncode := url.QueryEscape(credentials.Password)

  // Set the URL to clone
  fullUrlEncode := "https://" + userEncode + ":" + passwordEncode + "@" + urlRepo

  /*
    Execute the clone command.
    I'll explain this function next.
  */
  executeCloneCommand(fullUrlEncode, *cloneProjectName)

Before creating the function we will create a Struct with the structure of a user and Userswhat is an array of many users.

// Users struct, Array of users.
type Users struct {
  Users []User `json:"users"`
}

// User struct, JSON with attributes.
type User struct {
  Alias    string `json:"alias"`
  Email    string `json:"email"`
  Password string `json:"password"`
}

Now I'll explain the function findCredentials

/*
  Create a function, this receives one parameter:
  * alias - String

  The function will return a User struct.
*/
func findCredentials(alias string) User {
  // Get the current user, the user who call the command.
  user, err := user.Current()
  // create the credentials file path.
  credentialsPath := user.HomeDir + "/.codecommit/" + "credentials.json"
  // Open the credentials file.
  jsonFile, err := os.Open(credentialsPath)
  // Validate if hasn't error opening the file.
  if err != nil {
    log.Fatalln(err)
  }

  defer jsonFile.Close()
  // Read the file.
  byteValue, _ := ioutil.ReadAll(jsonFile)

  // Set a variable using the Users Struct.
  var users Users

  json.Unmarshal(byteValue, &users)

  // Make a for to find the right alias in the file.
  for i := 0; i < len(users.Users); i++ {
    if alias == users.Users[i].Alias {
      return users.Users[i]
    }
  }

  // If no exist the profile, send an error indicating what to do for add a new profile.
  fmt.Println("No profile with the value:", alias)
  log.Fatalln("Please add the user/password/alias in the credentials.json file inside" + user.HomeDir + "/.codecommit folder")
  return User{}
}

The last step is executeCloneCommand and I'll explain how to do this below

/*
  Create a function, this receives two parameter:
  * URL - String
  * projectName - String
*/
func executeCloneCommand(url string, projectName string) {
  /*
    Validate if the parameter "projectName" is a empty to set the parameter and execute the comand `git clone` else execute the gommand `git clone`without this patameter.
  */
  if projectName != "" {
    /*
      Execute the command with the project name, this will execute a command like this:
      `git clone https://url test`

      If executing the command has an error print it in console an exit the program.
    */
    if err := exec.Command("git", "clone", url, projectName).Run(); err != nil {
      fmt.Fprintln(os.Stderr, err)
      os.Exit(1)
    }
  } else {
    /*
      Execute the command without project name, this will execute a command like this:
      `git clone https://url test`

      If executing the command has an error print it in console an exit the program.
    */
    if err := exec.Command("git", "clone", url).Run(); err != nil {
      fmt.Fprintln(os.Stderr, err)
      os.Exit(1)
    }
  }
}

After writing all this you can test the command using go run main.go clone --profile="your-profile" --url="https://url" --projectName="test" and clone any project you have in AWS CodeCommit.

However if you want install the command go to root project and execute go install and now can clone from AWS CodeCommit using cloneaws clone --profile="your-profile" --url="https://url" --projectName="test"