2FA and OTP

  private String getCurrentCode(AccountIndex account, byte[] challenge) throws OtpSourceException {
    // Account is required.
    if (account == null) {
      throw new OtpSourceException("No account");
    }


    OtpType type = mAccountDb.getType(account);
    String secret = getSecret(account);


    long otpState = 0;


    if (type == OtpType.TOTP) {
      // For time-based OTP, the state is derived from clock.
      otpState =
          mTotpCounter.getValueAtTime(Utilities.millisToSeconds(mTotpClock.nowMillis()));
    } else if (type == OtpType.HOTP){
      // For counter-based OTP, the state is obtained by incrementing stored counter.
      mAccountDb.incrementCounter(account);
      Integer counter = mAccountDb.getCounter(account);
      otpState = counter.longValue();
    }


    String result = computePin(secret, otpState, challenge);
    return result;
  }

  @Override
  public long nowMillis() {
    return mSystemWallClock.nowMillis() + getTimeCorrectionMinutes() * Utilities.MINUTE_IN_MILLIS;
  }
  
  /**
   * Gets the currently used time correction value.
   * 
   * @return number of minutes by which this device is behind the correct time.
   */
  public int getTimeCorrectionMinutes() {
    synchronized (mLock) {
      if (mCachedCorrectionMinutes == null) {
        mCachedCorrectionMinutes = mPreferences.getInt(PREFERENCE_KEY_OFFSET_MINUTES, 0); 
      }
      return mCachedCorrectionMinutes;
    }
  }

  /**
   * Gets the value of this counter at the specified time.
   * 
   * @param time time instant (seconds since UNIX epoch) for which to obtain the value.
   * 
   * @return value of the counter at the {@code time}.
   */
  public long getValueAtTime(long time) {
    assertValidTime(time);
    
    // According to the RFC:
    // T = (Current Unix time - T0) / X, where the default floor function is used. 
    //   T  - counter value,
    //   T0 - start time.
    //   X  - time step.
    
    // It's important to use a floor function instead of simple integer division. For example,
    // assuming a time step of 3:
    // Time since start time: -6 -5 -4 -3 -2 -1  0  1  2  3  4  5  6
    // Correct value:         -2 -2 -2 -1 -1 -1  0  0  0  1  1  1  2
    // Simple division / 3:   -2 -1 -1 -1  0  0  0  0  0  1  1  1  2
    //
    // To avoid using Math.floor which requires imprecise floating-point arithmetic, we
    // we compute the value using integer division, but using a different equation for
    // negative and non-negative time since start time.
    long timeSinceStartTime = time - mStartTime;
    if (timeSinceStartTime >= 0) {
      return timeSinceStartTime / mTimeStep;
    } else {
      return (timeSinceStartTime - (mTimeStep - 1)) / mTimeStep;
    }
  }

  /**
   * Computes the one-time PIN given the secret key.
   *
   * @param secret the secret key
   * @param otpState current token state (counter or time-interval)
   * @param challenge optional challenge bytes to include when computing passcode.
   * @return the PIN
   */
  private String computePin(String secret, long otpState, byte[] challenge)
      throws OtpSourceException {
    if (secret == null || secret.length() == 0) {
      throw new OtpSourceException("Null or empty secret");
    }


    try {
      Signer signer = AccountDb.getSigningOracle(secret);  // [1]
      PasscodeGenerator pcg = new PasscodeGenerator(signer,  // [2]
        (challenge == null) ? PIN_LENGTH : REFLECTIVE_PIN_LENGTH);


      return (challenge == null) ?
             pcg.generateResponseCode(otpState) :
             pcg.generateResponseCode(otpState, challenge);
    } catch (GeneralSecurityException e) {
      throw new OtpSourceException("Crypto failure", e);
    }
  }

  public static Signer getSigningOracle(String secret) {
    try {
      byte[] keyBytes = decodeKey(secret);
      final Mac mac = Mac.getInstance("HMACSHA1");
      mac.init(new SecretKeySpec(keyBytes, ""));


      // Create a signer object out of the standard Java MAC implementation.
      return new Signer() {
        @Override
        public byte[] sign(byte[] data) {
          return mac.doFinal(data);
        }
      };
    } catch (DecodingException
        | NoSuchAlgorithmException
        | InvalidKeyException
        | IllegalArgumentException error) {
      Log.e(LOCAL_TAG, error.getMessage());
    }


    return null;
  }


  private static byte[] decodeKey(String secret) throws DecodingException {
    return Base32String.decode(secret);
  }

  /**
   * Powers of 10 to shorten the pin to the desired number of digits. This
   * prevents invalid OTP generation when Math.pow() is implemented incorrectly
   * (e.g. when 10^6 != 1000000), and matches the reference implementation in
   * RFC 6238. Must be kept in sync with {@link #MAX_PASSCODE_LENGTH}.
   */
  private static final int[] DIGITS_POWER =
      {1, 10, 100, 1000, 10000, 100000, 1000000, 10000000, 100000000, 1000000000};java
  
  /**
   * @param state 8-byte integer value representing internal OTP state.
   * @param challenge Optional challenge as array of bytes.
   * @return A decimal response code
   * @throws GeneralSecurityException If a JCE exception occur
   */
  public String generateResponseCode(long state, byte[] challenge)
      throws GeneralSecurityException {
    if (challenge == null) {
      return generateResponseCode(state);
    } else {
      // Allocate space for combination and store.
      byte[] value =
          ByteBuffer.allocate(8 + challenge.length)
              .putLong(state) // Write out OTP state
              .put(challenge, 0, challenge.length) // Concatenate with challenge.
              .array();
      return generateResponseCode(value);
    }
  }
  
  /**
   * @param challenge An arbitrary byte array used as a challenge
   * @return A decimal response code
   * @throws GeneralSecurityException If a JCE exception occur
   */
  public String generateResponseCode(byte[] challenge)
      throws GeneralSecurityException {
    byte[] hash = signer.sign(challenge);


    // Dynamically truncate the hash
    // OffsetBits are the low order bits of the last byte of the hash
    int offset = hash[hash.length - 1] & 0xF;
    // Grab a positive integer value starting at the given offset.
    int truncatedHash = hashToInt(hash, offset) & 0x7FFFFFFF;
    int pinValue = truncatedHash % DIGITS_POWER[codeLength];
    return padOutput(pinValue);
  }


  private String padOutput(int value) {
    String result = Integer.toString(value);
    for (int i = result.length(); i < codeLength; i++) {
      result = "0" + result;
    }
    return result;
  }

  /**
   * Grabs a positive integer value from the input array starting at
   * the given offset.
   * @param bytes the array of bytes
   * @param start the index into the array to start grabbing bytes
   * @return the integer constructed from the four bytes in the array
   */
  private int hashToInt(byte[] bytes, int start) {
    DataInput input = new DataInputStream(
        new ByteArrayInputStream(bytes, start, bytes.length - start));
    int val;
    try {
      val = input.readInt();
    } catch (IOException e) {
      throw new IllegalStateException(e);
    }
    return val;
  }

import sys
import hmac
import hashlib
import base64
import time
import struct

import unittest
from unittest.mock import patch


class TOTPGenerator:
    """
    TOTP Generator
    """

    def __init__(self, secret, interval=60):
        self.secret = secret
        self.interval = interval
        self.key_bytes = self.decode_key(secret)

    def decode_key(self, secret):
        return base64.b32decode(secret, casefold=True)

    def hmac_sha1(self, message):
        return hmac.new(self.key_bytes, message, hashlib.sha1).digest()

    def dynamic_truncation(self, hmac_result):
        offset = hmac_result[-1] & 0x0F
        truncated_hash = (
            struct.unpack(">I", hmac_result[offset : offset + 4])[0] & 0x7FFFFFFF
        )
        return truncated_hash

    def get_state(self):
        return int(time.time()) // self.interval

    def generate(self):
        state = self.get_state()

        hmac_result = self.hmac_sha1(struct.pack(">Q", state))

        truncated_hash = self.dynamic_truncation(hmac_result)

        pin_value = truncated_hash % (1_000_000)
        return self.pad_output(pin_value)

    def pad_output(self, value):
        result = str(value)
        return result.zfill(6)


class TestTOTPGenerator(unittest.TestCase):
    def setUp(self):
        self.secret = "7777777777777777"

    @patch.object(TOTPGenerator, "get_state", return_value=0)
    def test_generate_first_case(self, mock_get_state):
        _ = mock_get_state
        generator = TOTPGenerator(self.secret)
        result = generator.generate()
        self.assertEqual(result, "724477")

    @patch.object(TOTPGenerator, "get_state", return_value=123456789123456789)
    def test_generate_second_case(self, mock_get_state):
        _ = mock_get_state
        generator = TOTPGenerator(self.secret)
        result = generator.generate()
        self.assertEqual(result, "815107")


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--test":
        sys.argv.pop(1)
        unittest.main()

    secret = "JBQW4YLBONQWO2I="
    totp_generator = TOTPGenerator(secret)
    totp = totp_generator.generate()
    print("Current TOTP:", totp)